Struct google_cloud_spanner::transaction_rw::ReadWriteTransaction

pub struct ReadWriteTransaction { /* private fields */ }

ReadWriteTransaction provides a locking read-write transaction.

This type of transaction is the only way to write data into Cloud Spanner; Client::apply, Client::apply_at_least_once, Client::partitioned_update use transactions internally. These transactions rely on pessimistic locking and, if necessary, two-phase commit. Locking read-write transactions may abort, requiring the application to retry. However, the interface exposed by Client:run_with_retry eliminates the need for applications to write retry loops explicitly.

Locking transactions may be used to atomically read-modify-write data anywhere in a database. This type of transaction is externally consistent.

Clients should attempt to minimize the amount of time a transaction is active. Faster transactions commit with higher probability and cause less contention. Cloud Spanner attempts to keep read locks active as long as the transaction continues to do reads. Long periods of inactivity at the client may cause Cloud Spanner to release a transaction’s locks and abort it.

Reads performed within a transaction acquire locks on the data being read. Writes can only be done at commit time, after all reads have been completed. Conceptually, a read-write transaction consists of zero or more reads or SQL queries followed by a commit.

See Client::run_with_retry for an example.

Semantics

Cloud Spanner can commit the transaction if all read locks it acquired are still valid at commit time, and it is able to acquire write locks for all writes. Cloud Spanner can abort the transaction for any reason. If a commit attempt returns ABORTED, Cloud Spanner guarantees that the transaction has not modified any user data in Cloud Spanner.

Unless the transaction commits, Cloud Spanner makes no guarantees about how long the transaction’s locks were held for. It is an error to use Cloud Spanner locks for any sort of mutual exclusion other than between Cloud Spanner transactions themselves.

Aborted transactions

Application code does not need to retry explicitly; RunInTransaction will automatically retry a transaction if an attempt results in an abort. The lock priority of a transaction increases after each prior aborted transaction, meaning that the next attempt has a slightly better chance of success than before.

Under some circumstances (e.g., many transactions attempting to modify the same row(s)), a transaction can abort many times in a short period before successfully committing. Thus, it is not a good idea to cap the number of retries a transaction can attempt; instead, it is better to limit the total amount of wall time spent retrying.

Implementations

impl ReadWriteTransaction

pub async fn begin(
    session: ManagedSession,
    options: CallOptions
) -> Result<ReadWriteTransaction, BeginError>

Examples found in repository

src/client.rs (line 617)

    pub async fn begin_read_write_transaction(&self) -> Result<ReadWriteTransaction, TxError> {
        let session = self.get_session().await?;
        ReadWriteTransaction::begin(session, ReadWriteTransactionOption::default().begin_options)
            .await
            .map_err(|e| e.status.into())
    }

    /// Get open session count.
    pub fn session_count(&self) -> usize {
        self.sessions.num_opened()
    }

    async fn read_write_transaction_sync_with_option<T, E>(
        &self,
        f: impl Fn(&mut ReadWriteTransaction, Option<CancellationToken>) -> Result<T, E>,
        options: ReadWriteTransactionOption,
    ) -> Result<(Option<Timestamp>, T), E>
    where
        E: TryAs<Status> + From<SessionError> + From<Status>,
    {
        let (bo, co) = Client::split_read_write_transaction_option(options);

        let ro = TransactionRetrySetting::default();
        let session = Some(self.get_session().await?);

        // reuse session
        let cancel = bo.cancel.clone();
        invoke_fn(
            cancel.clone(),
            Some(ro),
            |session| async {
                let cancel = cancel.clone().map(|v| v.child_token());
                let mut tx = self.create_read_write_transaction::<E>(session, bo.clone()).await?;
                let result = f(&mut tx, cancel);
                tx.finish(result, Some(co.clone())).await
            },
            session,
        )
        .await
    }

    async fn create_read_write_transaction<E>(
        &self,
        session: Option<ManagedSession>,
        bo: CallOptions,
    ) -> Result<ReadWriteTransaction, (E, Option<ManagedSession>)>
    where
        E: TryAs<Status> + From<SessionError> + From<Status>,
    {
        ReadWriteTransaction::begin(session.unwrap(), bo)
            .await
            .map_err(|e| (E::from(e.status), Some(e.session)))
    }

pub async fn begin_partitioned_dml(
    session: ManagedSession,
    options: CallOptions
) -> Result<ReadWriteTransaction, BeginError>

Examples found in repository

src/client.rs (line 347)

    pub async fn partitioned_update_with_option(
        &self,
        stmt: Statement,
        options: PartitionedUpdateOption,
    ) -> Result<i64, TxError> {
        let ro = TransactionRetrySetting::new(vec![Code::Aborted, Code::Internal]);
        let session = Some(self.get_session().await?);

        // reuse session
        invoke_fn(
            options.begin_options.cancel.clone(),
            Some(ro),
            |session| async {
                let mut tx =
                    match ReadWriteTransaction::begin_partitioned_dml(session.unwrap(), options.begin_options.clone())
                        .await
                    {
                        Ok(tx) => tx,
                        Err(e) => return Err((TxError::GRPC(e.status), Some(e.session))),
                    };
                let qo = match options.query_options.clone() {
                    Some(o) => o,
                    None => QueryOptions::default(),
                };
                tx.update_with_option(stmt.clone(), qo)
                    .await
                    .map_err(|e| (TxError::GRPC(e), tx.take_session()))
            },
            session,
        )
        .await
    }

pub fn buffer_write(&mut self, ms: Vec<Mutation>)

Examples found in repository

src/client.rs (line 445)

    pub async fn apply_with_option(
        &self,
        ms: Vec<Mutation>,
        options: ReadWriteTransactionOption,
    ) -> Result<Option<Timestamp>, TxError> {
        let result: Result<(Option<Timestamp>, ()), TxError> = self
            .read_write_transaction_sync_with_option(
                |tx, _cancel| {
                    tx.buffer_write(ms.to_vec());
                    Ok(())
                },
                options,
            )
            .await;
        Ok(result?.0)
    }

pub async fn update(&mut self, stmt: Statement) -> Result<i64, Status>

pub async fn update_with_option(
    &mut self,
    stmt: Statement,
    options: QueryOptions
) -> Result<i64, Status>

Examples found in repository

src/transaction_rw.rs (line 165)

    pub async fn update(&mut self, stmt: Statement) -> Result<i64, Status> {
        self.update_with_option(stmt, QueryOptions::default()).await
    }

src/client.rs (line 357)

    pub async fn partitioned_update_with_option(
        &self,
        stmt: Statement,
        options: PartitionedUpdateOption,
    ) -> Result<i64, TxError> {
        let ro = TransactionRetrySetting::new(vec![Code::Aborted, Code::Internal]);
        let session = Some(self.get_session().await?);

        // reuse session
        invoke_fn(
            options.begin_options.cancel.clone(),
            Some(ro),
            |session| async {
                let mut tx =
                    match ReadWriteTransaction::begin_partitioned_dml(session.unwrap(), options.begin_options.clone())
                        .await
                    {
                        Ok(tx) => tx,
                        Err(e) => return Err((TxError::GRPC(e.status), Some(e.session))),
                    };
                let qo = match options.query_options.clone() {
                    Some(o) => o,
                    None => QueryOptions::default(),
                };
                tx.update_with_option(stmt.clone(), qo)
                    .await
                    .map_err(|e| (TxError::GRPC(e), tx.take_session()))
            },
            session,
        )
        .await
    }

pub async fn batch_update(
    &mut self,
    stmt: Vec<Statement>
) -> Result<Vec<i64>, Status>

pub async fn batch_update_with_option(
    &mut self,
    stmt: Vec<Statement>,
    options: QueryOptions
) -> Result<Vec<i64>, Status>

Examples found in repository

src/transaction_rw.rs (line 193)

    pub async fn batch_update(&mut self, stmt: Vec<Statement>) -> Result<Vec<i64>, Status> {
        self.batch_update_with_option(stmt, QueryOptions::default()).await
    }

pub async fn end<S, E>(
    &mut self,
    result: Result<S, E>,
    options: Option<CommitOptions>
) -> Result<(Option<Timestamp>, S), E>where
    E: TryAs<Status> + From<Status>,

Methods from Deref<Target = Transaction>

pub async fn query(
    &mut self,
    statement: Statement
) -> Result<RowIterator<'_>, Status>

query executes a query against the database. It returns a RowIterator for retrieving the resulting rows.

query returns only row data, without a query plan or execution statistics.

pub async fn query_with_option(
    &mut self,
    statement: Statement,
    options: QueryOptions
) -> Result<RowIterator<'_>, Status>

query executes a query against the database. It returns a RowIterator for retrieving the resulting rows.

query returns only row data, without a query plan or execution statistics.

Examples found in repository

src/transaction.rs (line 90)

    pub async fn query(&mut self, statement: Statement) -> Result<RowIterator<'_>, Status> {
        self.query_with_option(statement, QueryOptions::default()).await
    }

pub async fn read(
    &mut self,
    table: &str,
    columns: &[&str],
    key_set: impl Into<KeySet>
) -> Result<RowIterator<'_>, Status>

read returns a RowIterator for reading multiple rows from the database.

use google_cloud_spanner::key::Key;
use google_cloud_spanner::client::Client;
use google_cloud_spanner::reader::AsyncIterator;

#[tokio::main]
async fn main() -> Result<(), anyhow::Error> {
    const DATABASE: &str = "projects/local-project/instances/test-instance/databases/local-database";
    let client = Client::new(DATABASE).await?;

    let mut tx = client.single().await?;
    let mut iter = tx.read("Guild", &["GuildID", "OwnerUserID"], vec![
        Key::new(&"pk1"),
        Key::new(&"pk2")
    ]).await?;

    while let Some(row) = iter.next().await? {
        let guild_id = row.column_by_name::<String>("GuildID");
        //do something
    };
    Ok(())
}

pub async fn read_with_option(
    &mut self,
    table: &str,
    columns: &[&str],
    key_set: impl Into<KeySet>,
    options: ReadOptions
) -> Result<RowIterator<'_>, Status>

read returns a RowIterator for reading multiple rows from the database.

Examples found in repository

src/transaction.rs (line 152)

    pub async fn read(
        &mut self,
        table: &str,
        columns: &[&str],
        key_set: impl Into<KeySet>,
    ) -> Result<RowIterator<'_>, Status> {
        self.read_with_option(table, columns, key_set, ReadOptions::default())
            .await
    }

    /// read returns a RowIterator for reading multiple rows from the database.
    pub async fn read_with_option(
        &mut self,
        table: &str,
        columns: &[&str],
        key_set: impl Into<KeySet>,
        options: ReadOptions,
    ) -> Result<RowIterator<'_>, Status> {
        let request = ReadRequest {
            session: self.get_session_name(),
            transaction: Some(self.transaction_selector.clone()),
            table: table.to_string(),
            index: options.index,
            columns: columns.iter().map(|x| x.to_string()).collect(),
            key_set: Some(key_set.into().inner),
            limit: options.limit,
            resume_token: vec![],
            partition_token: vec![],
            request_options: Transaction::create_request_options(options.call_options.priority),
        };

        let session = self.as_mut_session();
        let reader = Box::new(TableReader { request });
        RowIterator::new(session, reader, Some(options.call_options)).await
    }

    /// read returns a RowIterator for reading multiple rows from the database.
    /// ```
    /// use google_cloud_spanner::key::Key;
    /// use google_cloud_spanner::client::Client;
    ///
    /// #[tokio::main]
    /// async fn main() -> Result<(), anyhow::Error> {
    ///     const DATABASE: &str = "projects/local-project/instances/test-instance/databases/local-database";
    ///     let client = Client::new(DATABASE).await?;
    ///     let mut tx = client.single().await?;
    ///     let row = tx.read_row("Guild", &["GuildID", "OwnerUserID"], Key::new(&"guild1")).await?;
    ///     Ok(())
    /// }
    /// ```
    pub async fn read_row(&mut self, table: &str, columns: &[&str], key: Key) -> Result<Option<Row>, Status> {
        self.read_row_with_option(table, columns, key, ReadOptions::default())
            .await
    }

    /// read returns a RowIterator for reading multiple rows from the database.
    pub async fn read_row_with_option(
        &mut self,
        table: &str,
        columns: &[&str],
        key: Key,
        options: ReadOptions,
    ) -> Result<Option<Row>, Status> {
        let call_options = options.call_options.clone();
        let mut reader = self
            .read_with_option(table, columns, KeySet::from(key), options)
            .await?;
        reader.set_call_options(call_options);
        reader.next().await
    }

pub async fn read_row(
    &mut self,
    table: &str,
    columns: &[&str],
    key: Key
) -> Result<Option<Row>, Status>

read returns a RowIterator for reading multiple rows from the database.

use google_cloud_spanner::key::Key;
use google_cloud_spanner::client::Client;

#[tokio::main]
async fn main() -> Result<(), anyhow::Error> {
    const DATABASE: &str = "projects/local-project/instances/test-instance/databases/local-database";
    let client = Client::new(DATABASE).await?;
    let mut tx = client.single().await?;
    let row = tx.read_row("Guild", &["GuildID", "OwnerUserID"], Key::new(&"guild1")).await?;
    Ok(())
}

pub async fn read_row_with_option(
    &mut self,
    table: &str,
    columns: &[&str],
    key: Key,
    options: ReadOptions
) -> Result<Option<Row>, Status>

read returns a RowIterator for reading multiple rows from the database.

Examples found in repository

src/transaction.rs (line 197)

    pub async fn read_row(&mut self, table: &str, columns: &[&str], key: Key) -> Result<Option<Row>, Status> {
        self.read_row_with_option(table, columns, key, ReadOptions::default())
            .await
    }

Trait Implementations

impl Deref for ReadWriteTransaction

type Target = Transaction

The resulting type after dereferencing.

fn deref(&self) -> &Self::Target

Dereferences the value.

impl DerefMut for ReadWriteTransaction

fn deref_mut(&mut self) -> &mut Transaction

Mutably dereferences the value.