事务概览

本页面介绍了 Spanner 中的事务,并介绍了 Spanner 的读写、只读和分区 DML 事务接口。

Spanner 中的事务是一组读写操作。事务中的所有操作都是原子的,这意味着它们要么全部成功,要么全部失败。

会话用于在 Spanner 数据库中执行事务。会话代表一种与 Spanner 数据库服务的逻辑通信渠道。会话一次可以执行一个或多个事务。如需了解详情,请参阅会话。

事务类型

Spanner 支持以下事务类型,每种类型均针对特定的数据互动模式而设计:

  • 读写:这些事务用于读取和写入操作,这些操作后面还伴随提交操作。它们可能会获取锁定。如果获取失败,则需要重试。虽然它们仅限于单个数据库,但可以在该数据库中跨多个表修改数据。

  • 只读:这些事务可保证在多次读取操作中数据的一致性,但不允许修改数据。它们会在系统确定的时间戳(为了一致性)或用户配置的过去时间戳执行。与读写事务不同,它们不需要提交操作或锁定。不过,它们可能会暂停以等待正在进行的写入操作结束。

  • 分区 DML:此事务类型将 DML 语句作为分区 DML 操作来执行。它经过优化,可大规模执行 DML 语句,但施加了一些限制,以确保该语句是幂等的,并且可分区,使得每个分区能够独立于其他分区执行。对于不需要原子事务的大量写入,请考虑使用批量写入。如需了解详情,请参阅使用批量写入修改数据。

读写事务

读写事务由零个或多个读取或查询语句(后跟提交请求)组成。在提交请求之前的任何时间,客户端都可以发送回滚请求来中止事务。

可序列化隔离

使用默认的可序列化隔离级别,读写事务可以原子方式读取、修改和写入数据。这种类型的事务具有外部一致性。

使用读写事务时,我们建议您尽量缩短事务的活跃时间。事务时长越短,锁定时间就越短,成功提交的可能性就越大,争用情况就越少。这是因为长时间锁定可能会导致死锁和事务中止。只要事务继续执行读取操作且未通过提交或回滚终止,Spanner 就会尝试让读取锁定保持活跃状态。如果客户端长时间处于非活跃状态,Spanner 可能会释放事务锁并中止该事务。

如需执行依赖于一项或多项读取操作的写入操作,请使用读写事务:

  • 如果您必须以原子方式提交一项或多项写入操作,请在同一读写事务内执行这些写入。例如,如果您将 200 美元从账号 A 转到账号 B,请在同一事务中执行两项写入操作(将账号 A 减少 200 美元,并将账号 B 增加 200 美元)和初始账号余额读取。
  • 如果您想将账号 A 的余额翻倍,请在同一事务内执行读取和写入操作。这可确保系统在余额翻倍和更新之前读取余额。
  • 如果写入操作依赖于读取操作,请在同一读写事务中执行这两项操作,即使写入操作未执行也是如此。例如,如果您希望仅在账号 A 的余额大于 500 美元时将 200 美元从账号 A 转到账号 B,请在同一事务内添加对 A 余额的读取操作和条件写入操作,即使转账操作未发生也是如此。

如需执行读取操作,请使用单次读取方法或只读事务:

  • 如果您只执行读取操作,并且可以使用单次读取方法来表示读取操作,请使用该单次读取方法或只读事务。与读写事务不同,单次读取不会获取锁定。

可重复读隔离

在 Spanner 中,可重复读隔离是使用一种称为快照隔离的技术实现的。可重复读隔离可确保事务内的所有读取操作与事务开始时的数据库保持一致。它还保证,只有在没有冲突的情况下,对同一数据的并发写入才会成功。

在默认的乐观锁定模式下,如果需要写入数据,则在提交之前不会获取任何锁定。如果写入的数据发生冲突,或者由于 Spanner 内的暂时性事件(例如服务器重启)导致冲突,Spanner 仍可能会中止事务。由于读写事务中的读取操作在可重复读隔离中不会获取锁定,因此在只读事务或读写事务中执行只读操作没有任何区别。

遇到以下情况时,请考虑在可重复读隔离中使用读写事务:

  • 工作负载是读取密集型,且写入冲突较少。
  • 应用遇到性能瓶颈,原因是锁争用导致延迟,以及较旧的高优先级事务伤害较新的低优先级事务以防止潜在的死锁(伤害-等待),从而导致事务中止。
  • 应用不需要可序列化隔离级别提供的更严格的保证。

在执行写入操作且该操作依赖于一项或多项读取操作时,在可重复读隔离下可能会出现写偏差。写偏差是由一种特定的并发更新引起的,虽然可以单独接受每种更新,但它们组合的效果违反了应用数据完整性。因此,请务必使用 FOR UPDATE 子句或 lock_scanned_ranges=exclusive 提示来执行属于事务关键部分的读取操作,以避免写偏差。如需了解详情,请参阅读写冲突和正确性,以及读写语义中讨论的示例。

接口

Spanner 客户端库提供了一个接口,用于在读写事务内执行操作,并在事务中止的情况下重试。事务可能需要多次重试才能提交。

有几种情况可能会导致事务中止。例如,如果两项事务尝试同时修改数据,则可能会发生死锁。在此类情况下,Spanner 会中止一项事务,以让另一项事务可以继续进行。更少见的情况是,Spanner 内的暂时性事件也可能导致事务中止。

所有的读写事务都具备关系型数据库的 ACID 属性。由于事务具有原子性,因此中止的事务不会影响数据库。Spanner 客户端库会自动重试此类事务,但如果您不使用客户端库,请在同一会话内重试事务,以提高成功率。每次导致 ABORTED 错误的重试都会提高事务的锁定优先级。此外,Spanner 客户端驱动程序还包含内部事务重试逻辑,可通过重新运行事务来屏蔽暂时性错误。

在 Spanner 客户端库中使用事务时,您可以将事务的主体定义为函数对象。此函数封装了对一个或多个数据库表执行的读写操作。Spanner 客户端库会反复执行此函数,直到事务成功提交或遇到无法重试的错误。

示例

假设您在 Albums 表中有一个 MarketingBudget 列:

CREATE TABLE Albums (
  SingerId        INT64 NOT NULL,
  AlbumId         INT64 NOT NULL,
  AlbumTitle      STRING(MAX),
  MarketingBudget INT64
) PRIMARY KEY (SingerId, AlbumId);

您的营销部门要求您将 20 万美元从 Albums (2, 2) 的预算划拨给 Albums (1, 1),但前提是该专辑的预算资金充裕。您应该为此操作使用锁定读写事务,因为该事务可能会根据读取结果执行写入。

以下客户端库示例展示了如何使用默认的可序列化隔离级别执行读写事务:

C++

void ReadWriteTransaction(google::cloud::spanner::Client client) {
  namespace spanner = ::google::cloud::spanner;
  using ::google::cloud::Status;
  using ::google::cloud::StatusCode;
  using ::google::cloud::StatusOr;

  // A helper to read a single album MarketingBudget.
  auto get_current_budget =
      [](spanner::Client client, spanner::Transaction txn,
         std::int64_t singer_id,
         std::int64_t album_id) -> StatusOr<std::int64_t> {
    auto key = spanner::KeySet().AddKey(spanner::MakeKey(singer_id, album_id));
    auto rows = client.Read(std::move(txn), "Albums", std::move(key),
                            {"MarketingBudget"});
    using RowType = std::tuple<std::int64_t>;
    auto row = spanner::GetSingularRow(spanner::StreamOf<RowType>(rows));
    if (!row) return std::move(row).status();
    return std::get<0>(*std::move(row));
  };

  auto constexpr kInsufficientFundsMessage =
      "The second album does not have enough funds to transfer";
  auto commit = client.Commit(
      [&](spanner::Transaction const& txn) -> StatusOr<spanner::Mutations> {
        auto b1 = get_current_budget(client, txn, 1, 1);
        if (!b1) return std::move(b1).status();
        auto b2 = get_current_budget(client, txn, 2, 2);
        if (!b2) return std::move(b2).status();
        std::int64_t transfer_amount = 200000;

        if (*b2 < transfer_amount) {
          return Status(StatusCode::kFailedPrecondition,
                        kInsufficientFundsMessage);
        }

        return spanner::Mutations{
            spanner::UpdateMutationBuilder(
                "Albums", {"SingerId", "AlbumId", "MarketingBudget"})
                .EmplaceRow(1, 1, *b1 + transfer_amount)
                .EmplaceRow(2, 2, *b2 - transfer_amount)
                .Build()};
      });

  if (!commit) throw std::move(commit).status();
  std::cout << "Transfer was successful [spanner_read_write_transaction]\n";
}

C#


using Google.Cloud.Spanner.Data;
using System;
using System.Threading.Tasks;
using System.Transactions;

public class ReadWriteWithTransactionAsyncSample
{
    public async Task<int> ReadWriteWithTransactionAsync(string projectId, string instanceId, string databaseId)
    {
        // This sample transfers 200,000 from the MarketingBudget
        // field of the second Album to the first Album. Make sure to run
        // the Add Column and Write Data To New Column samples first,
        // in that order.

        string connectionString = $"Data Source=projects/{projectId}/instances/{instanceId}/databases/{databaseId}";

        using TransactionScope scope = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled);
        decimal transferAmount = 200000;
        decimal secondBudget = 0;
        decimal firstBudget = 0;

        using var connection = new SpannerConnection(connectionString);
        using var cmdLookup1 = connection.CreateSelectCommand("SELECT * FROM Albums WHERE SingerId = 2 AND AlbumId = 2");

        using (var reader = await cmdLookup1.ExecuteReaderAsync())
        {
            while (await reader.ReadAsync())
            {
                // Read the second album's budget.
                secondBudget = reader.GetFieldValue<decimal>("MarketingBudget");
                // Confirm second Album's budget is sufficient and
                // if not raise an exception. Raising an exception
                // will automatically roll back the transaction.
                if (secondBudget < transferAmount)
                {
                    throw new Exception($"The second album's budget {secondBudget} is less than the amount to transfer.");
                }
            }
        }

        // Read the first album's budget.
        using var cmdLookup2 = connection.CreateSelectCommand("SELECT * FROM Albums WHERE SingerId = 1 and AlbumId = 1");
        using (var reader = await cmdLookup2.ExecuteReaderAsync())
        {
            while (await reader.ReadAsync())
            {
                firstBudget = reader.GetFieldValue<decimal>("MarketingBudget");
            }
        }

        // Specify update command parameters.
        using var cmdUpdate = connection.CreateUpdateCommand("Albums", new SpannerParameterCollection
        {
            { "SingerId", SpannerDbType.Int64 },
            { "AlbumId", SpannerDbType.Int64 },
            { "MarketingBudget", SpannerDbType.Int64 },
        });

        // Update second album to remove the transfer amount.
        secondBudget -= transferAmount;
        cmdUpdate.Parameters["SingerId"].Value = 2;
        cmdUpdate.Parameters["AlbumId"].Value = 2;
        cmdUpdate.Parameters["MarketingBudget"].Value = secondBudget;
        var rowCount = await cmdUpdate.ExecuteNonQueryAsync();

        // Update first album to add the transfer amount.
        firstBudget += transferAmount;
        cmdUpdate.Parameters["SingerId"].Value = 1;
        cmdUpdate.Parameters["AlbumId"].Value = 1;
        cmdUpdate.Parameters["MarketingBudget"].Value = firstBudget;
        rowCount += await cmdUpdate.ExecuteNonQueryAsync();
        scope.Complete();
        Console.WriteLine("Transaction complete.");
        return rowCount;
    }
}

Go


import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/spanner"
)

func writeWithTransaction(w io.Writer, db string) error {
	ctx := context.Background()
	client, err := spanner.NewClient(ctx, db)
	if err != nil {
		return err
	}
	defer client.Close()

	_, err = client.ReadWriteTransaction(ctx, func(ctx context.Context, txn *spanner.ReadWriteTransaction) error {
		getBudget := func(key spanner.Key) (int64, error) {
			row, err := txn.ReadRow(ctx, "Albums", key, []string{"MarketingBudget"})
			if err != nil {
				return 0, err
			}
			var budget int64
			if err := row.Column(0, &budget); err != nil {
				return 0, err
			}
			return budget, nil
		}
		album2Budget, err := getBudget(spanner.Key{2, 2})
		if err != nil {
			return err
		}
		const transferAmt = 200000
		if album2Budget >= transferAmt {
			album1Budget, err := getBudget(spanner.Key{1, 1})
			if err != nil {
				return err
			}
			album1Budget += transferAmt
			album2Budget -= transferAmt
			cols := []string{"SingerId", "AlbumId", "MarketingBudget"}
			txn.BufferWrite([]*spanner.Mutation{
				spanner.Update("Albums", cols, []interface{}{1, 1, album1Budget}),
				spanner.Update("Albums", cols, []interface{}{2, 2, album2Budget}),
			})
			fmt.Fprintf(w, "Moved %d from Album2's MarketingBudget to Album1's.", transferAmt)
		}
		return nil
	})
	return err
}

Java

static void writeWithTransaction(DatabaseClient dbClient) {
  dbClient
      .readWriteTransaction()
      .run(transaction -> {
        // Transfer marketing budget from one album to another. We do it in a transaction to
        // ensure that the transfer is atomic.
        Struct row =
            transaction.readRow("Albums", Key.of(2, 2), Arrays.asList("MarketingBudget"));
        long album2Budget = row.getLong(0);
        // Transaction will only be committed if this condition still holds at the time of
        // commit. Otherwise it will be aborted and the callable will be rerun by the
        // client library.
        long transfer = 200000;
        if (album2Budget >= transfer) {
          long album1Budget =
              transaction
                  .readRow("Albums", Key.of(1, 1), Arrays.asList("MarketingBudget"))
                  .getLong(0);
          album1Budget += transfer;
          album2Budget -= transfer;
          transaction.buffer(
              Mutation.newUpdateBuilder("Albums")
                  .set("SingerId")
                  .to(1)
                  .set("AlbumId")
                  .to(1)
                  .set("MarketingBudget")
                  .to(album1Budget)
                  .build());
          transaction.buffer(
              Mutation.newUpdateBuilder("Albums")
                  .set("SingerId")
                  .to(2)
                  .set("AlbumId")
                  .to(2)
                  .set("MarketingBudget")
                  .to(album2Budget)
                  .build());
        }
        return null;
      });
}

Node.js

// This sample transfers 200,000 from the MarketingBudget field
// of the second Album to the first Album, as long as the second
// Album has enough money in its budget. Make sure to run the
// addColumn and updateData samples first (in that order).

// Imports the Google Cloud client library
const {Spanner} = require('@google-cloud/spanner');

/**
 * TODO(developer): Uncomment the following lines before running the sample.
 */
// const projectId = 'my-project-id';
// const instanceId = 'my-instance';
// const databaseId = 'my-database';

// Creates a client
const spanner = new Spanner({
  projectId: projectId,
});

// Gets a reference to a Cloud Spanner instance and database
const instance = spanner.instance(instanceId);
const database = instance.database(databaseId);

const transferAmount = 200000;

// Note: the `runTransaction()` method is non blocking and returns "void".
// For sequential execution of the transaction use `runTransactionAsync()` method which returns a promise.
// For example: await database.runTransactionAsync(async (err, transaction) => { ... })
database.runTransaction(async (err, transaction) => {
  if (err) {
    console.error(err);
    return;
  }
  let firstBudget, secondBudget;
  const queryOne = {
    columns: ['MarketingBudget'],
    keys: [[2, 2]], // SingerId: 2, AlbumId: 2
  };

  const queryTwo = {
    columns: ['MarketingBudget'],
    keys: [[1, 1]], // SingerId: 1, AlbumId: 1
  };

  Promise.all([
    // Reads the second album's budget
    transaction.read('Albums', queryOne).then(results => {
      // Gets second album's budget
      const rows = results[0].map(row => row.toJSON());
      secondBudget = rows[0].MarketingBudget;
      console.log(`The second album's marketing budget: ${secondBudget}`);

      // Makes sure the second album's budget is large enough
      if (secondBudget < transferAmount) {
        throw new Error(
          `The second album's budget (${secondBudget}) is less than the transfer amount (${transferAmount}).`,
        );
      }
    }),

    // Reads the first album's budget
    transaction.read('Albums', queryTwo).then(results => {
      // Gets first album's budget
      const rows = results[0].map(row => row.toJSON());
      firstBudget = rows[0].MarketingBudget;
      console.log(`The first album's marketing budget: ${firstBudget}`);
    }),
  ])
    .then(() => {
      console.log(firstBudget, secondBudget);
      // Transfers the budgets between the albums
      firstBudget += transferAmount;
      secondBudget -= transferAmount;

      console.log(firstBudget, secondBudget);

      // Updates the database
      // Note: Cloud Spanner interprets Node.js numbers as FLOAT64s, so they
      // must be converted (back) to strings before being inserted as INT64s.
      transaction.update('Albums', [
        {
          SingerId: '1',
          AlbumId: '1',
          MarketingBudget: firstBudget.toString(),
        },
        {
          SingerId: '2',
          AlbumId: '2',
          MarketingBudget: secondBudget.toString(),
        },
      ]);
    })
    .then(() => {
      // Commits the transaction and send the changes to the database
      return transaction.commit();
    })
    .then(() => {
      console.log(
        `Successfully executed read-write transaction to transfer ${transferAmount} from Album 2 to Album 1.`,
      );
    })
    .catch(err => {
      console.error('ERROR:', err);
    })
    .then(() => {
      transaction.end();
      // Closes the database when finished
      return database.close();
    });
});

PHP

use Google\Cloud\Spanner\SpannerClient;
use Google\Cloud\Spanner\Transaction;
use UnexpectedValueException;

/**
 * Performs a read-write transaction to update two sample records in the
 * database.
 *
 * This will transfer 200,000 from the `MarketingBudget` field for the second
 * Album to the first Album. If the `MarketingBudget` for the second Album is
 * too low, it will raise an exception.
 *
 * Before running this sample, you will need to run the `update_data` sample
 * to populate the fields.
 * Example:
 * ```
 * read_write_transaction($instanceId, $databaseId);
 * ```
 *
 * @param string $instanceId The Spanner instance ID.
 * @param string $databaseId The Spanner database ID.
 */
function read_write_transaction(string $instanceId, string $databaseId): void
{
    $spanner = new SpannerClient();
    $instance = $spanner->instance($instanceId);
    $database = $instance->database($databaseId);

    $database->runTransaction(function (Transaction $t) use ($spanner) {
        $transferAmount = 200000;

        // Read the second album's budget.
        $secondAlbumKey = [2, 2];
        $secondAlbumKeySet = $spanner->keySet(['keys' => [$secondAlbumKey]]);
        $secondAlbumResult = $t->read(
            'Albums',
            $secondAlbumKeySet,
            ['MarketingBudget'],
            ['limit' => 1]
        );

        $firstRow = $secondAlbumResult->rows()->current();
        $secondAlbumBudget = $firstRow['MarketingBudget'];
        if ($secondAlbumBudget < $transferAmount) {
            // Throwing an exception will automatically roll back the transaction.
            throw new UnexpectedValueException(
                'The second album\'s budget is lower than the transfer amount: ' . $transferAmount
            );
        }

        $firstAlbumKey = [1, 1];
        $firstAlbumKeySet = $spanner->keySet(['keys' => [$firstAlbumKey]]);
        $firstAlbumResult = $t->read(
            'Albums',
            $firstAlbumKeySet,
            ['MarketingBudget'],
            ['limit' => 1]
        );

        // Read the first album's budget.
        $firstRow = $firstAlbumResult->rows()->current();
        $firstAlbumBudget = $firstRow['MarketingBudget'];

        // Update the budgets.
        $secondAlbumBudget -= $transferAmount;
        $firstAlbumBudget += $transferAmount;
        printf('Setting first album\'s budget to %s and the second album\'s ' .
            'budget to %s.' . PHP_EOL, $firstAlbumBudget, $secondAlbumBudget);

        // Update the rows.
        $t->updateBatch('Albums', [
            ['SingerId' => 1, 'AlbumId' => 1, 'MarketingBudget' => $firstAlbumBudget],
            ['SingerId' => 2, 'AlbumId' => 2, 'MarketingBudget' => $secondAlbumBudget],
        ]);

        // Commit the transaction!
        $t->commit();

        print('Transaction complete.' . PHP_EOL);
    });
}

Python

def read_write_transaction(instance_id, database_id):
    """Performs a read-write transaction to update two sample records in the
    database.

    This will transfer 200,000 from the `MarketingBudget` field for the second
    Album to the first Album. If the `MarketingBudget` is too low, it will
    raise an exception.

    Before running this sample, you will need to run the `update_data` sample
    to populate the fields.
    """
    spanner_client = spanner.Client()
    instance = spanner_client.instance(instance_id)
    database = instance.database(database_id)

    def update_albums(transaction):
        # Read the second album budget.
        second_album_keyset = spanner.KeySet(keys=[(2, 2)])
        second_album_result = transaction.read(
            table="Albums",
            columns=("MarketingBudget",),
            keyset=second_album_keyset,
            limit=1,
        )
        second_album_row = list(second_album_result)[0]
        second_album_budget = second_album_row[0]

        transfer_amount = 200000

        if second_album_budget < transfer_amount:
            # Raising an exception will automatically roll back the
            # transaction.
            raise ValueError("The second album doesn't have enough funds to transfer")

        # Read the first album's budget.
        first_album_keyset = spanner.KeySet(keys=[(1, 1)])
        first_album_result = transaction.read(
            table="Albums",
            columns=("MarketingBudget",),
            keyset=first_album_keyset,
            limit=1,
        )
        first_album_row = list(first_album_result)[0]
        first_album_budget = first_album_row[0]

        # Update the budgets.
        second_album_budget -= transfer_amount
        first_album_budget += transfer_amount
        print(
            "Setting first album's budget to {} and the second album's "
            "budget to {}.".format(first_album_budget, second_album_budget)
        )

        # Update the rows.
        transaction.update(
            table="Albums",
            columns=("SingerId", "AlbumId", "MarketingBudget"),
            values=[(1, 1, first_album_budget), (2, 2, second_album_budget)],
        )

    database.run_in_transaction(update_albums)

    print("Transaction complete.")

Ruby

# project_id  = "Your Google Cloud project ID"
# instance_id = "Your Spanner instance ID"
# database_id = "Your Spanner database ID"

require "google/cloud/spanner"

spanner         = Google::Cloud::Spanner.new project: project_id
client          = spanner.client instance_id, database_id
transfer_amount = 200_000

client.transaction do |transaction|
  first_album  = transaction.read("Albums", [:MarketingBudget], keys: [[1, 1]]).rows.first
  second_album = transaction.read("Albums", [:MarketingBudget], keys: [[2, 2]]).rows.first

  raise "The second album does not have enough funds to transfer" if second_album[:MarketingBudget] < transfer_amount

  new_first_album_budget  = first_album[:MarketingBudget] + transfer_amount
  new_second_album_budget = second_album[:MarketingBudget] - transfer_amount

  transaction.update "Albums", [
    { SingerId: 1, AlbumId: 1, MarketingBudget: new_first_album_budget  },
    { SingerId: 2, AlbumId: 2, MarketingBudget: new_second_album_budget }
  ]
end

puts "Transaction complete"

如需查看有关如何���用可重复读隔离执行读写事务的示例,请参阅使用可重复读隔离级别。

语义

本部分介绍了 Spanner 中读写事务的语义。

属性

可序列化隔离是 Spanner 中的默认隔离级别。在可序列化隔离下,Spanner 为客户端提供最严格的事务并发控制保证,即外部一致性。读写事务以原子方式执行一组读写操作。写入操作可以继续进行,而不会被只读事务阻塞。读写事务执行时的时间戳与经过的时间相匹配。序列化顺序与此时间戳顺序一致。

得益于 Cloud Spanner 的这些特性,应用开发者可以专注于每项事务本身的正确性,而不必想方设法让其执行不受可能同时执行的其他事务干扰。

您还可以使用可重复读隔离级别执行读写事务。可重复读隔离可确保事务内的所有读取操作都能看到事务开始时数据库的一致强快照。如需了解详情,请参阅可重复读隔离。

采用可序列化隔离的读写事务

在默认的串行化隔离级别下,成功提交一个包含一系列读写操作的事务后,以下条件适用:

  • 事务返回的值可反映事务提交时间戳的一致快照。
  • 空的行或范围在提交时保持为空。
  • 事务在事务的提交时间戳提交所有写入。
  • 在事务提交之前,任何事务都看不到写入。

Spanner 客户端驱动程序包含事务重试逻辑,以遮盖暂时性错误,方法是重新运行事务并验证客户端观察到的数据。

其结果是所有读写操作似乎是在 (无论是从事务本身还是从事务 并分析 Spanner 数据库的其他读取者和写入者的角度。 这意味着读取和写入发生在同一时间戳。如需查看示例,请参阅可序列化和外部一致性。

采用可重复读隔离的读写事务

在成功提交采用可重复读隔离的事务后,以下条件适用:

  • 该事务返回反映数据库一致快照的值。该快照通常在事务的���一个操作期间建立,可能与提交时间戳不同。
  • 由于可重复读是使用快照隔离实现的,因此事务只有在写入集在事务快照时间戳和提交时间戳之间没有发生变化的情况下,才会在事务的提交时间戳进行所有写入的提交。
  • 其他事务只有在该事务提交后才能看到这些写入操作。

读写事务中只读操作的隔离

当读写事务仅执行读取操作时,它会提供与只读事务类似的一致性保证。事务内的所有读取都会返回来自一致时间戳的数据,包括确认不存在的行。

一个区别是,当读写事务在未执行写入操作的情况下提交时。在这种情况下,无法保证在读取操作和事务提交之间,事务内读取的数据在数据库中保持不变。

为了确保数据新鲜度,并验证数据自上次检索后未修改,需要进行后续读取。此重读可以在另一个读写事务内执行,也可以通过强读取执行。

为了达到最佳效率,如果事务专门执行读取,请使用只读事务,而不是读写事务,尤其是在使用可序列化隔离时。

可序列化和外部一致性与可重复读有何不同

默认情况下,Spanner 提供强大的事务保证,包括可序列化和外部一致性。这些特性确保数据保持一致,并且操作按照可预测的顺序发生,即使在分布式环境中也是如此。

可序列化可确保所有事务看起来都是按单一、连续的顺序依次执行,即使它们是同时处理也是如此。Spanner 通过为事务分配提交时间戳来实现这一点,反映事务的提交顺序。

Spanner 提供更强的保证,称为外部一致性。这意味着,事务不仅按提交时间戳反映的顺序进行提交,而且这些时间戳也与实际时间一致。这样一来,您就可以将提交时间戳与实时时间进行比较,从而以全球一致的顺序查看数据。

从本质上讲,如果事务 Txn1 实时地在另一个事务 Txn2 之前提交,则 Txn1 的提交时间戳会早于 Txn2 的提交时间戳。

请参考以下示例:

显示读取相同数据的两项事务的执行时间轴

在此场景中,在时间轴 t 期间:

  • 事务 Txn1 读取数据 A,暂存写入 A,然后成功提交。
  • 事务 Txn2 在 Txn1 启动后开始。它会读取数据 B,然后读取数据 A。

即使 Txn2 在 Txn1 完成之前就已启动,Txn2 也会观察 Txn1 对 A 所做的更改。这是因为 Txn2 在 Txn1 提交对 A 的写入后读取 A。

虽然 Txn1 和 Txn2 的执行时间可能会重叠,但它们的提交时间戳(分别为 c1 和 c2)会强制执行线性事务顺序。这意味着:

  • Txn1 内的所有读取和写入似乎都发生在单一时间点 c1。
  • Txn2 内的所有读取和写入似乎都发生在单一时间点 c2。
  • 重要的是,对于已提交的写入,c1 比 c2 更早,即使写入发生在不同的机器上也是如此。如果 Txn2 仅执行读取,则 c1 会在 c2 之前或与其同时执行。

这种强有序性意味着,如果后续读取操作观察到 Txn2 的影响,则也会观察到 Txn1 的影响。对于所有成功提交的事务,此属性都为 true。

另一方面,如果您使用可重复读隔离,相同的事务可能会发生以下情况:

  • Txn1 首先读取数据 A,并在当时创建自己的数据库快照。
  • Txn2 随后开始,读取数据 B,并建立自己的快照。
  • 接下来,Txn1 修改数据 A,并成功提交其更改。
  • Txn2 尝试读取数据 A。关键在于,由于 Txn2 是在较早的快照上操作,因此它看不到 Txn1 刚刚对 A 所做的更新。Txn2 读取的是旧值。
  • Txn2 修改数据 B 并提交。

在这种情况下,每个事务都在自己的数据库一致快照上操作,该快照是在事务开始时获取的。如果 Txn2 对 B 的写入在逻辑上依赖于其从 A 读取的值,则此序列可能会导致写入偏差异常。本质上,Txn2 基于过时的信息进行了更新,其后续写入可能会违反应用级不变量约束。为防止出现这种情况,请考虑在可重复读隔离中使用 SELECT...FOR UPDATE,或在架构中创建检查限制条件。

事务失败时的读取和写入保证

如果执行事务的调用失败,那么您所获得的读取和写入保证取决于底层的提交调用是因什么错误而失败。

Spanner 可能会在内部多次执行事务操作。如果执行尝试失败,返回的错误会指明所发生的情况,从而指明您所获得的保证。不过,如果 Spanner 重试您的事务,则其操作产生的任何副作用(例如,更改外部系统或 Spanner 数据库之外的系统状态)可能会多次发生。

当 Spanner 事务失败时,您��读取和写入所获得的保证取决于提交操作期间遇到的特定错误。

例如,“未找到行”或“行已存在”等错误消息表示在写入缓冲的数据变更时遇到了问题。例如,如果客户端尝试更新的行不存在,则可能会出现这种情况。在以下情况下:

  • 读取是一致的:在事务期间读取的任何数据均保证在出现错误之前是一致的。
  • 未应用写入:事务尝试的变更未提交到数据库。
  • 行一致性:触发错误的行不存在(或存在状态)与在事务内执行的读取一致。

您可以随时取消 Spanner 中的异步读取操作,而不会影响同一事务内的其他正在进行的操作。如果更高级别的操作已取消,或者您决定根据初始结果中止读取,这种灵活性会非常有用。

不过,请务必了解,请求取消读取并不保证立即终止。在取消请求后,读取操作仍可能会:

  • 成功完成:读取可能会在取消生效之前完成处理并返回结果。
  • 因其他原因失败:读取可能会因其他错误(例如取消)而终止。
  • 返回不完整的结果:读取可能会返回部分结果,然后作为事务提交流程的一部分进行验证。

取消提交操作会中止整个事务,除非事务已提交或因其他原因而失败。

原子性、一致性、持久性

除了隔离之外,Spanner 还提供其他 ACID 属性保证:

  • 原子性:如果一个事务的所有操作要么都成功完成,要么一个都没有完成,那么该事务就被认为是原子的。如果事务内的任何操作失败,整个事务都会回滚到原始状态,以确保数据完整性。
  • 一致性:事务必须维护数据库规则和约束条件的完整性。事务完成后,数据库应处于有效状态,并遵循预定义的规则。
  • 持久性:事务提交后,其更改将永久存储在数据库中,即使发生系统故障、断电或其他中断,更改内容也会持久存在。

性能

本部分介绍了会影响读写事务性能的问题。

锁定并发控制

默认情况下,Spanner 允许多个客户端在其默认的串行化隔离级别下并发地与同一个数据库进行交互。为了维护这些并发事务之间的数据一致性,Spanner 有一个同时使用共享锁和独占锁的锁定机制。只能为串行化事务获取这些读取锁,而不能为采用可重复读隔离的事务获取。

当一个串行化事务执行读取操作时,Spanner 会获取相关数据的共享读取锁。这些共享锁允许其他并发读取操作访问相同的数据。这种并发性会一直保持,直到您的事务准备提交其更改。

在可序列化隔离中,在提交阶段,随着写入操作的应用,事务会尝试将其锁定升级为独占锁定。为此,Spanner 会执行以下操作:

  • 阻止对受影响的数据发出任何新的共享读取锁定请求。
  • 等待释放对该数据的所有现有共享读取锁定。
  • 在清除所有共享读取锁定后,它会施加独占锁定,在写入期间授予其对数据的独占访问权限。

在可重复读隔离中提交事务时,事务会获取所写入数据的独占锁定。如果并发事务也提交同一数据的写入,则该事务可能必须等待锁定。

有关锁定的注意事项:

  • 粒度:Spanner 会在行和列的粒度上应用锁定。这意味着,如果事务 T1 持有行 albumid 的列 A 的锁定,事务 T2 仍然可以同时写入同一行 albumid 的列 B 而不会发生冲突。
  • 不需要读取的写入:

    • 如果事务中没有读取操作,Spanner 可能无需对不需要读取的写入使用独占锁定。它可以使用写入者共享锁定。这是因为,对于不需要读取的写入,其应用顺序由提交时间戳决定,从而允许多个写入者可以同时对同一项进行操作而不会发生冲突。只有当您的事务首先读取要写入的数据时,才需要独占锁定。
    • 在可重复读隔离中,事务通常会在提交时为写入的单元格获取独占锁定。
  • 用于行查询的二级索引:在可序列化隔离中,当在读写事务内执行读取操作时,使用二级索引可以显著提高性能。通过使用二级索引将扫描的行限制在较小的范围内,Spanner 锁定表中较少的行,从而允许对该特定范围之外的行进行更大的并发修改。

  • 外部资源独占访问:Spanner 的内部锁定旨在确保 Spanner 数据库本身的数据一致性。请勿使用它们来确保对 Spanner 外部的资源进行独占访问。Spanner 可能会出于多种原因而取消事务,包括内部系统优化(例如跨计算资源的数据移动)。如果重试事务(无论是通过应用代码明确进行,还是通过 Spanner JDBC 驱动程序等客户端库隐式进行),则只能保证在成功提交的尝试期间持有锁定。

  • 锁定统计信息:如需诊断和调查数据库内的锁定冲突,请使用锁定统计信息内省工具。

死锁检测

Spanner 会检测多项事务可能导致死锁的情况,并强制除一项事务以外的所有其他事务取消。设想以下场景:Txn1 持有记录 A 的锁定,并正在等待记录 B 的锁定,而 Txn2 持有记录 B 的锁定并正在等待记录 A 的锁定。为了解决此问题,必须取消其中一个事务,释放其锁定,并允许另一个事务继续进行。

Spanner 使用标准的“受伤-等待”算法来检测死锁。在后台,Spanner 会跟踪请求冲突锁定的每个事务的存在时间,并允许较早的事务取消较晚的事务。较早的事务是指最早的读取、查询或提交发生的时间更早的事务。

通过优先处理较早的事务,Spanner 可确保每项事务最终都有机会获取锁定,只要其存在时间足够长,使其优先级高于其他事务。例如,需要写入者共享锁定的较早事务可以取消持有读取者共享锁定的较晚事务。

分布式执行

Spanner 可以对跨多台服务器的数据执行事务,但与单服务器事务相比,此功能需要付出性能代价。

哪些类型的事务可能是分布式的?Spanner 可以将数据库行的责任分摊到多个服务器。通常,某行及其对应的交错表行由同一服务器处理,就像同一个表中键邻近的两行。Spanner 可以跨不同服务器上的行执行事务。然而,一般来说,相比那些涉及分散在整个数据库或大型表中的许多行的事务,只涉及相邻的许多行的事务通常执行速度更快、开销更小。

Spanner 中最高效的事务仅包括应以原子方式应用的读取和写入。当所有读取和写入操作均访问位于键空间同一部分的数据时,事务处理速度最快。

只读事务

除了锁定读写事务外,Spanner 还提供只读事务。

当您需要在同一时间戳执行多项读取时,请使用只读事务。如果您可以使用 Spanner 的某个单次读取方法来表达您的读取,则应该使用该单次读取方法。使用这种单次读取调用的性能应该与在只读事务中执行单次读取的性能相当。

如果要读取大量数据,请考虑使用分区来并行读取数据。

由于只读事务不会写入,它们不会持有锁定,也不会阻止其他事务。只读事务会观察到一致的事务提交历史记录前缀,因此您的应用始终可获得一致的数据。

接口

Spanner 提供了一个接口,用于在只读事务的情境中执行操作,并在事务中止的情况下重试。

示例

以下示例展示了如何使用只读事务,为同一时间戳的两次读取获取一致的数据:

C++

void ReadOnlyTransaction(google::cloud::spanner::Client client) {
  namespace spanner = ::google::cloud::spanner;
  auto read_only = spanner::MakeReadOnlyTransaction();

  spanner::SqlStatement select(
      "SELECT SingerId, AlbumId, AlbumTitle FROM Albums");
  using RowType = std::tuple<std::int64_t, std::int64_t, std::string>;

  // Read#1.
  auto rows1 = client.ExecuteQuery(read_only, select);
  std::cout << "Read 1 results\n";
  for (auto& row : spanner::StreamOf<RowType>(rows1)) {
    if (!row) throw std::move(row).status();
    std::cout << "SingerId: " << std::get<0>(*row)
              << " AlbumId: " << std::get<1>(*row)
              << " AlbumTitle: " << std::get<2>(*row) << "\n";
  }
  // Read#2. Even if changes occur in-between the reads the transaction ensures
  // that Read #1 and Read #2 return the same data.
  auto rows2 = client.ExecuteQuery(read_only, select);
  std::cout << "Read 2 results\n";
  for (auto& row : spanner::StreamOf<RowType>(rows2)) {
    if (!row) throw std::move(row).status();
    std::cout << "SingerId: " << std::get<0>(*row)
              << " AlbumId: " << std::get<1>(*row)
              << " AlbumTitle: " << std::get<2>(*row) << "\n";
  }
}

C#


using Google.Cloud.Spanner.Data;
using System;
using System.Collections.Generic;
using System.Threading.Tasks;
using System.Transactions;

public class QueryDataWithTransactionAsyncSample
{
    public class Album
    {
        public int SingerId { get; set; }
        public int AlbumId { get; set; }
        public string AlbumTitle { get; set; }
    }

    public async Task<List<Album>> QueryDataWithTransactionAsync(string projectId, string instanceId, string databaseId)
    {
        string connectionString = $"Data Source=projects/{projectId}/instances/{instanceId}/databases/{databaseId}";

        var albums = new List<Album>();
        using TransactionScope scope = new TransactionScope(TransactionScopeAsyncFlowOption.Enabled);
        using var connection = new SpannerConnection(connectionString);

        // Opens the connection so that the Spanner transaction included in the TransactionScope
        // is read-only TimestampBound.Strong.
        await connection.OpenAsync(SpannerTransactionCreationOptions.ReadOnly, options: null, cancellationToken: default);
        using var cmd = connection.CreateSelectCommand("SELECT SingerId, AlbumId, AlbumTitle FROM Albums");

        // Read #1.
        using (var reader = await cmd.ExecuteReaderAsync())
        {
            while (await reader.ReadAsync())
            {
                Console.WriteLine("SingerId : " + reader.GetFieldValue<string>("SingerId")
                    + " AlbumId : " + reader.GetFieldValue<string>("AlbumId")
                    + " AlbumTitle : " + reader.GetFieldValue<string>("AlbumTitle"));
            }
        }

        // Read #2. Even if changes occur in-between the reads,
        // the transaction ensures that Read #1 and Read #2
        // return the same data.
        using (var reader = await cmd.ExecuteReaderAsync())
        {
            while (await reader.ReadAsync())
            {
                albums.Add(new Album
                {
                    AlbumId = reader.GetFieldValue<int>("AlbumId"),
                    SingerId = reader.GetFieldValue<int>("SingerId"),
                    AlbumTitle = reader.GetFieldValue<string>("AlbumTitle")
                });
            }
        }
        scope.Complete();
        Console.WriteLine("Transaction complete.");
        return albums;
    }
}

Go


import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/spanner"
	"google.golang.org/api/iterator"
)

func readOnlyTransaction(w io.Writer, db string) error {
	ctx := context.Background()
	client, err := spanner.NewClient(ctx, db)
	if err != nil {
		return err
	}
	defer client.Close()

	ro := client.ReadOnlyTransaction()
	defer ro.Close()
	stmt := spanner.Statement{SQL: `SELECT SingerId, AlbumId, AlbumTitle FROM Albums`}
	iter := ro.Query(ctx, stmt)
	defer iter.Stop()
	for {
		row, err := iter.Next()
		if err == iterator.Done {
			break
		}
		if err != nil {
			return err
		}
		var singerID int64
		var albumID int64
		var albumTitle string
		if err := row.Columns(&singerID, &albumID, &albumTitle); err != nil {
			return err
		}
		fmt.Fprintf(w, "%d %d %s\n", singerID, albumID, albumTitle)
	}

	iter = ro.Read(ctx, "Albums", spanner.AllKeys(), []string{"SingerId", "AlbumId", "AlbumTitle"})
	defer iter.Stop()
	for {
		row, err := iter.Next()
		if err == iterator.Done {
			return nil
		}
		if err != nil {
			return err
		}
		var singerID int64
		var albumID int64
		var albumTitle string
		if err := row.Columns(&singerID, &albumID, &albumTitle); err != nil {
			return err
		}
		fmt.Fprintf(w, "%d %d %s\n", singerID, albumID, albumTitle)
	}
}

Java

static void readOnlyTransaction(DatabaseClient dbClient) {
  // ReadOnlyTransaction must be closed by calling close() on it to release resources held by it.
  // We use a try-with-resource block to automatically do so.
  try (ReadOnlyTransaction transaction = dbClient.readOnlyTransaction()) {
    try (ResultSet queryResultSet =
        transaction.executeQuery(
            Statement.of("SELECT SingerId, AlbumId, AlbumTitle FROM Albums"))) {
      while (queryResultSet.next()) {
        System.out.printf(
            "%d %d %s\n",
            queryResultSet.getLong(0), queryResultSet.getLong(1), queryResultSet.getString(2));
      }
    } // queryResultSet.close() is automatically called here
    try (ResultSet readResultSet =
        transaction.read(
          "Albums", KeySet.all(), Arrays.asList("SingerId", "AlbumId", "AlbumTitle"))) {
      while (readResultSet.next()) {
        System.out.printf(
            "%d %d %s\n",
            readResultSet.getLong(0), readResultSet.getLong(1), readResultSet.getString(2));
      }
    } // readResultSet.close() is automatically called here
  } // transaction.close() is automatically called here
}

Node.js

// Imports the Google Cloud client library
const {Spanner} = require('@google-cloud/spanner');

/**
 * TODO(developer): Uncomment the following lines before running the sample.
 */
// const projectId = 'my-project-id';
// const instanceId = 'my-instance';
// const databaseId = 'my-database';

// Creates a client
const spanner = new Spanner({
  projectId: projectId,
});

// Gets a reference to a Cloud Spanner instance and database
const instance = spanner.instance(instanceId);
const database = instance.database(databaseId);

// Gets a transaction object that captures the database state
// at a specific point in time
database.getSnapshot(async (err, transaction) => {
  if (err) {
    console.error(err);
    return;
  }
  const queryOne = 'SELECT SingerId, AlbumId, AlbumTitle FROM Albums';

  try {
    // Read #1, using SQL
    const [qOneRows] = await transaction.run(queryOne);

    qOneRows.forEach(row => {
      const json = row.toJSON();
      console.log(
        `SingerId: ${json.SingerId}, AlbumId: ${json.AlbumId}, AlbumTitle: ${json.AlbumTitle}`,
      );
    });

    const queryTwo = {
      columns: ['SingerId', 'AlbumId', 'AlbumTitle'],
    };

    // Read #2, using the `read` method. Even if changes occur
    // in-between the reads, the transaction ensures that both
    // return the same data.
    const [qTwoRows] = await transaction.read('Albums', queryTwo);

    qTwoRows.forEach(row => {
      const json = row.toJSON();
      console.log(
        `SingerId: ${json.SingerId}, AlbumId: ${json.AlbumId}, AlbumTitle: ${json.AlbumTitle}`,
      );
    });

    console.log('Successfully executed read-only transaction.');
  } catch (err) {
    console.error('ERROR:', err);
  } finally {
    transaction.end();
    // Close the database when finished.
    await database.close();
  }
});

PHP

use Google\Cloud\Spanner\SpannerClient;

/**
 * Reads data inside of a read-only transaction.
 *
 * Within the read-only transaction, or "snapshot", the application sees
 * consistent view of the database at a particular timestamp.
 * Example:
 * ```
 * read_only_transaction($instanceId, $databaseId);
 * ```
 *
 * @param string $instanceId The Spanner instance ID.
 * @param string $databaseId The Spanner database ID.
 */
function read_only_transaction(string $instanceId, string $databaseId): void
{
    $spanner = new SpannerClient();
    $instance = $spanner->instance($instanceId);
    $database = $instance->database($databaseId);

    $snapshot = $database->snapshot();
    $results = $snapshot->execute(
        'SELECT SingerId, AlbumId, AlbumTitle FROM Albums'
    );
    print('Results from the first read:' . PHP_EOL);
    foreach ($results as $row) {
        printf('SingerId: %s, AlbumId: %s, AlbumTitle: %s' . PHP_EOL,
            $row['SingerId'], $row['AlbumId'], $row['AlbumTitle']);
    }

    // Perform another read using the `read` method. Even if the data
    // is updated in-between the reads, the snapshot ensures that both
    // return the same data.
    $keySet = $spanner->keySet(['all' => true]);
    $results = $database->read(
        'Albums',
        $keySet,
        ['SingerId', 'AlbumId', 'AlbumTitle']
    );

    print('Results from the second read:' . PHP_EOL);
    foreach ($results->rows() as $row) {
        printf('SingerId: %s, AlbumId: %s, AlbumTitle: %s' . PHP_EOL,
            $row['SingerId'], $row['AlbumId'], $row['AlbumTitle']);
    }
}

Python

def read_only_transaction(instance_id, database_id):
    """Reads data inside of a read-only transaction.

    Within the read-only transaction, or "snapshot", the application sees
    consistent view of the database at a particular timestamp.
    """
    spanner_client = spanner.Client()
    instance = spanner_client.instance(instance_id)
    database = instance.database(database_id)

    with database.snapshot(multi_use=True) as snapshot:
        # Read using SQL.
        results = snapshot.execute_sql(
            "SELECT SingerId, AlbumId, AlbumTitle FROM Albums"
        )

        print("Results from first read:")
        for row in results:
            print("SingerId: {}, AlbumId: {}, AlbumTitle: {}".format(*row))

        # Perform another read using the `read` method. Even if the data
        # is updated in-between the reads, the snapshot ensures that both
        # return the same data.
        keyset = spanner.KeySet(all_=True)
        results = snapshot.read(
            table="Albums", columns=("SingerId", "AlbumId", "AlbumTitle"), keyset=keyset
        )

        print("Results from second read:")
        for row in results:
            print("SingerId: {}, AlbumId: {}, AlbumTitle: {}".format(*row))

Ruby

# project_id  = "Your Google Cloud project ID"
# instance_id = "Your Spanner instance ID"
# database_id = "Your Spanner database ID"

require "google/cloud/spanner"

spanner = Google::Cloud::Spanner.new project: project_id
client  = spanner.client instance_id, database_id

client.snapshot do |snapshot|
  snapshot.execute("SELECT SingerId, AlbumId, AlbumTitle FROM Albums").rows.each do |row|
    puts "#{row[:AlbumId]} #{row[:AlbumTitle]} #{row[:SingerId]}"
  end

  # Even if changes occur in-between the reads, the transaction ensures that
  # both return the same data.
  snapshot.read("Albums", [:AlbumId, :AlbumTitle, :SingerId]).rows.each do |row|
    puts "#{row[:AlbumId]} #{row[:AlbumTitle]} #{row[:SingerId]}"
  end
end

语义

本部分介绍了只读事务的语义。

快照只读事务

当只读事务在 Spanner 中执行时,它会在单个逻辑时间点执行所有读取。这意味着,只读事务和任何其他并发读取者和写入者都会在特定时刻看到数据库的一致快照。

与锁定读写事务相比,这些快照只读事务为一致读取提供了更简单的方法。原因如下:

  • 无锁:只读事务不会获取锁定。相反,它们通过选择 Spanner 时间戳,并针对该数据的历史版本执行所有读取来运行。由于它们不会使用锁定,因此不会阻止并发读写事务。
  • 无取消:这些事务永远不会取消。虽然如果其选择的读取时间戳被垃圾回收,它们可能会失败,但 Spanner 的默认垃圾回收策略通常足够宽松,因此大多数应用都不会遇到此问题。
  • 无提交或回滚:只读事务不需要调用 sessions.commit 或 sessions.rollback,并且实际上会被阻止这样做。

为了执行快照事务,客户端会定义时间戳边界,以指示 Spanner 如何选择读取时间戳。时间戳边界有以下几种类型:

  • 强读:这些读取可保证您在读取开始之前看到已提交的所有事务的影响。单次读取内的所有行都是一致的。不过,强读不可重复,尽管强读确实会返回时间戳,并且在同一时间戳再次读取是可重复的。由于并发写入,两个连续强只读事务可能会产生不同的结果。对变更数据流的查询必须使用此边界。如需了解详情,请参阅 TransactionOptions.ReadOnly.strong。
  • 精确过时:此选项会在您指定的时间戳(以绝对时间戳或相对于当前时间的过时时长表示)执行读取。它可确保您观察到该时间戳之前的全局事务历史记录的前缀一致,并阻止可能以小于或等于读取时间戳的时间戳提交的冲突事务。虽然比有界限过时模式稍快,但它可能会返回较旧的数据。如需了解详情,请参阅 TransactionOptions.ReadOnly.read_timestamp 和 TransactionOptions.ReadOnly.exact_staleness。
  • 有界限过时:Spanner 在用户定义的过时限制范围内选择最新的时间戳,允许在最近的可用副本上执行操作而不会阻塞。返回的所有行都是一致的。与强读取一样,有界限过时不可重复,因为不同的读取即使使用相同的边界,也可能会在不同的时间戳执行。这些读取分为两个阶段(先���行时间戳协商,然后进行读取),通常比精确过时读取稍慢,但它们通常会返回更新的结果,并且更有可能在本地副本上执行。此模式仅适用于一次性只读事务,因为时间戳协商需要事先知道要读取哪些行。如需了解详情,请参阅 TransactionOptions.ReadOnly.max_staleness 和 TransactionOptions.ReadOnly.min_read_timestamp。

分区 DML 事务

您可以使用分区 DML 执行大量 UPDATE 和 DELETE 语句,并且不会遇到事务限制或锁定整个表。Spanner 通过对键空间进行分区,并在单独的读写事务中对每个分区执行 DML 语句来实现这一点。

如需使用非分区 DML,您可以在代码中明确创建的读写事务中执行语句。如需了解详情,请参阅使用 DML。

接口

Spanner 提供了 TransactionOptions.partitionedDml 接口,用于执行单个分区 DML 语句。

示例

以下代码示例更新 Albums 表的 MarketingBudget 列。

C++

您可以使用 ExecutePartitionedDml() 函数来执行分区 DML 语句。

void DmlPartitionedUpdate(google::cloud::spanner::Client client) {
  namespace spanner = ::google::cloud::spanner;
  auto result = client.ExecutePartitionedDml(
      spanner::SqlStatement("UPDATE Albums SET MarketingBudget = 100000"
                            "  WHERE SingerId > 1"));
  if (!result) throw std::move(result).status();
  std::cout << "Updated at least " << result->row_count_lower_bound
            << " row(s) [spanner_dml_partitioned_update]\n";
}

C#

您可以使用 ExecutePartitionedUpdateAsync() 方法来执行分区 DML 语句。


using Google.Cloud.Spanner.Data;
using System;
using System.Threading.Tasks;

public class UpdateUsingPartitionedDmlCoreAsyncSample
{
    public async Task<long> UpdateUsingPartitionedDmlCoreAsync(string projectId, string instanceId, string databaseId)
    {
        string connectionString = $"Data Source=projects/{projectId}/instances/{instanceId}/databases/{databaseId}";

        using var connection = new SpannerConnection(connectionString);
        await connection.OpenAsync();

        using var cmd = connection.CreateDmlCommand("UPDATE Albums SET MarketingBudget = 100000 WHERE SingerId > 1");
        long rowCount = await cmd.ExecutePartitionedUpdateAsync();

        Console.WriteLine($"{rowCount} row(s) updated...");
        return rowCount;
    }
}

Go

您可以使用 PartitionedUpdate() 方法来执行分区 DML 语句。


import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/spanner"
)

func updateUsingPartitionedDML(w io.Writer, db string) error {
	ctx := context.Background()
	client, err := spanner.NewClient(ctx, db)
	if err != nil {
		return err
	}
	defer client.Close()

	stmt := spanner.Statement{SQL: "UPDATE Albums SET MarketingBudget = 100000 WHERE SingerId > 1"}
	rowCount, err := client.PartitionedUpdate(ctx, stmt)
	if err != nil {
		return err
	}
	fmt.Fprintf(w, "%d record(s) updated.\n", rowCount)
	return nil
}

Java

您可以使用 executePartitionedUpdate() 方法来执行分区 DML 语句。

static void updateUsingPartitionedDml(DatabaseClient dbClient) {
  String sql = "UPDATE Albums SET MarketingBudget = 100000 WHERE SingerId > 1";
  long rowCount = dbClient.executePartitionedUpdate(Statement.of(sql));
  System.out.printf("%d records updated.\n", rowCount);
}

Node.js

您可以使用 runPartitionedUpdate() 方法来执行分区 DML 语句。

// Imports the Google Cloud client library
const {Spanner} = require('@google-cloud/spanner');

/**
 * TODO(developer): Uncomment the following lines before running the sample.
 */
// const projectId = 'my-project-id';
// const instanceId = 'my-instance';
// const databaseId = 'my-database';

// Creates a client
const spanner = new Spanner({
  projectId: projectId,
});

// Gets a reference to a Cloud Spanner instance and database
const instance = spanner.instance(instanceId);
const database = instance.database(databaseId);

try {
  const [rowCount] = await database.runPartitionedUpdate({
    sql: 'UPDATE Albums SET MarketingBudget = 100000 WHERE SingerId > 1',
  });
  console.log(`Successfully updated ${rowCount} records.`);
} catch (err) {
  console.error('ERROR:', err);
} finally {
  // Close the database when finished.
  database.close();
}

PHP

您可以使用 executePartitionedUpdate() 方法来执行分区 DML 语句。

use Google\Cloud\Spanner\SpannerClient;

/**
 * Updates sample data in the database by partition with a DML statement.
 *
 * This updates the `MarketingBudget` column which must be created before
 * running this sample. You can add the column by running the `add_column`
 * sample or by running this DDL statement against your database:
 *
 *     ALTER TABLE Albums ADD COLUMN MarketingBudget INT64
 *
 * Example:
 * ```
 * update_data($instanceId, $databaseId);
 * ```
 *
 * @param string $instanceId The Spanner instance ID.
 * @param string $databaseId The Spanner database ID.
 */
function update_data_with_partitioned_dml(string $instanceId, string $databaseId): void
{
    $spanner = new SpannerClient();
    $instance = $spanner->instance($instanceId);
    $database = $instance->database($databaseId);

    $rowCount = $database->executePartitionedUpdate(
        'UPDATE Albums SET MarketingBudget = 100000 WHERE SingerId > 1'
    );

    printf('Updated %d row(s).' . PHP_EOL, $rowCount);
}

Python

您可以使用 execute_partitioned_dml() 方法来执行分区 DML 语句。

# instance_id = "your-spanner-instance"
# database_id = "your-spanner-db-id"

spanner_client = spanner.Client()
instance = spanner_client.instance(instance_id)
database = instance.database(database_id)

row_ct = database.execute_partitioned_dml(
    "UPDATE Albums SET MarketingBudget = 100000 WHERE SingerId > 1"
)

print("{} records updated.".format(row_ct))

Ruby

您可以使用 execute_partitioned_update() 方法来执行分区 DML 语句。

# project_id  = "Your Google Cloud project ID"
# instance_id = "Your Spanner instance ID"
# database_id = "Your Spanner database ID"

require "google/cloud/spanner"

spanner = Google::Cloud::Spanner.new project: project_id
client  = spanner.client instance_id, database_id

row_count = client.execute_partition_update(
  "UPDATE Albums SET MarketingBudget = 100000 WHERE SingerId > 1"
)

puts "#{row_count} records updated."

以下代码示例根据 SingerId 列从 Singers 表中删除行。

C++

void DmlPartitionedDelete(google::cloud::spanner::Client client) {
  namespace spanner = ::google::cloud::spanner;
  auto result = client.ExecutePartitionedDml(
      spanner::SqlStatement("DELETE FROM Singers WHERE SingerId > 10"));
  if (!result) throw std::move(result).status();
  std::cout << "Deleted at least " << result->row_count_lower_bound
            << " row(s) [spanner_dml_partitioned_delete]\n";
}

C#


using Google.Cloud.Spanner.Data;
using System;
using System.Threading.Tasks;

public class DeleteUsingPartitionedDmlCoreAsyncSample
{
    public async Task<long> DeleteUsingPartitionedDmlCoreAsync(string projectId, string instanceId, string databaseId)
    {
        string connectionString = $"Data Source=projects/{projectId}/instances/{instanceId}/databases/{databaseId}";

        using var connection = new SpannerConnection(connectionString);
        await connection.OpenAsync();

        using var cmd = connection.CreateDmlCommand("DELETE FROM Singers WHERE SingerId > 10");
        long rowCount = await cmd.ExecutePartitionedUpdateAsync();

        Console.WriteLine($"{rowCount} row(s) deleted...");
        return rowCount;
    }
}

Go


import (
	"context"
	"fmt"
	"io"

	"cloud.google.com/go/spanner"
)

func deleteUsingPartitionedDML(w io.Writer, db string) error {
	ctx := context.Background()
	client, err := spanner.NewClient(ctx, db)
	if err != nil {
		return err
	}
	defer client.Close()

	stmt := spanner.Statement{SQL: "DELETE FROM Singers WHERE SingerId > 10"}
	rowCount, err := client.PartitionedUpdate(ctx, stmt)
	if err != nil {
		return err

	}
	fmt.Fprintf(w, "%d record(s) deleted.", rowCount)
	return nil
}

Java

static void deleteUsingPartitionedDml(DatabaseClient dbClient) {
  String sql = "DELETE FROM Singers WHERE SingerId > 10";
  long rowCount = dbClient.executePartitionedUpdate(Statement.of(sql));
  System.out.printf("%d records deleted.\n", rowCount);
}

Node.js

// Imports the Google Cloud client library
const {Spanner} = require('@google-cloud/spanner');

/**
 * TODO(developer): Uncomment the following lines before running the sample.
 */
// const projectId = 'my-project-id';
// const instanceId = 'my-instance';
// const databaseId = 'my-database';

// Creates a client
const spanner = new Spanner({
  projectId: projectId,
});

// Gets a reference to a Cloud Spanner instance and database
const instance = spanner.instance(instanceId);
const database = instance.database(databaseId);

try {
  const [rowCount] = await database.runPartitionedUpdate({
    sql: 'DELETE FROM Singers WHERE SingerId > 10',
  });
  console.log(`Successfully deleted ${rowCount} records.`);
} catch (err) {
  console.error('ERROR:', err);
} finally {
  // Close the database when finished.
  database.close();
}

PHP

use Google\Cloud\Spanner\SpannerClient;

/**
 * Delete sample data in the database by partition with a DML statement.
 *
 * This updates the `MarketingBudget` column which must be created before
 * running this sample. You can add the column by running the `add_column`
 * sample or by running this DDL statement against your database:
 *
 *     ALTER TABLE Albums ADD COLUMN MarketingBudget INT64
 *
 * Example:
 * ```
 * update_data($instanceId, $databaseId);
 * ```
 *
 * @param string $instanceId The Spanner instance ID.
 * @param string $databaseId The Spanner database ID.
 */
function delete_data_with_partitioned_dml(string $instanceId, string $databaseId): void
{
    $spanner = new SpannerClient();
    $instance = $spanner->instance($instanceId);
    $database = $instance->database($databaseId);

    $rowCount = $database->executePartitionedUpdate(
        'DELETE FROM Singers WHERE SingerId > 10'
    );

    printf('Deleted %d row(s).' . PHP_EOL, $rowCount);
}

Python

# instance_id = "your-spanner-instance"
# database_id = "your-spanner-db-id"
spanner_client = spanner.Client()
instance = spanner_client.instance(instance_id)
database = instance.database(database_id)

row_ct = database.execute_partitioned_dml("DELETE FROM Singers WHERE SingerId > 10")

print("{} record(s) deleted.".format(row_ct))

Ruby

# project_id  = "Your Google Cloud project ID"
# instance_id = "Your Spanner instance ID"
# database_id = "Your Spanner database ID"

require "google/cloud/spanner"

spanner = Google::Cloud::Spanner.new project: project_id
client  = spanner.client instance_id, database_id

row_count = client.execute_partition_update(
  "DELETE FROM Singers WHERE SingerId > 10"
)

puts "#{row_count} records deleted."

语义

本部分介绍了分区 DML 的语义。

了解分区 DML 执行

无论是使用客户端库方法还是使用 Google Cloud CLI,一次只能执行一个分区 DML 语句。

分区事务不支持提交或回滚。Spanner 会��即执行并应用 DML 语句。如果您取消操作或操作失败,Spanner 将取消所有正在执行的分区,并且不会启动其余任何分区。不过,Spanner 不会回滚已执行的任何分区。

分区 DML 锁定获取策略

为减少锁定争用,分区 DML 仅对与 WHERE 子句匹配的行获取读取锁定。针对每个分区使用的较小的独立事务也会占用更少的时间来持有锁定。

旧读取时间戳和版本垃圾回收

Spanner 执行版本垃圾回收,以收集已删除或已覆盖的数据并回收存储空间。默认情况下,系统会回收超过一小时的数据。Spanner 无法在配置的 VERSION_RETENTION_PERIOD 之前的时间戳执行读取,默认值为 1 小时,但最多可配置为 1 周。如果读取在执行期间变得过旧,则会失败并返回 FAILED_PRECONDITION 错误。

对变更数据流的查询

变更数据流是一种架构对象,您可以将其配置为监控整个数据库、特定表或数据库内定义的一组列中的数据修改情况。

当您创建变更数据流时,Spanner 会定义一个对应的 SQL 表值函数 (TVF)。您可以使用此 TVF 通过 sessions.executeStreamingSql 方法查询关联变更数据流中的更改记录。TVF 的名称是根据变更数据流的名称生成的,并且始终以 READ_ 开头。

对变更数据流 TVF 的所有查询都必须在具有强烈只读 timestamp_bound 的一次性只读事务中使用 sessions.executeStreamingSql API 执行。借助变更数据流 TVF,您可以为时间范围指定 start_timestamp 和 end_timestamp。您可以使用此强只读 timestamp_bound 访问保留期限内的所有变更记录。所有其他 TransactionOptions 对变更数据流查询都无效。

此外,如果 TransactionOptions.read_only.return_read_timestamp 设置为 true,则描述事务的 Transaction 消息会返回 2^63 - 2 的特殊值,而不是有效的读取时间戳。您应舍弃此特殊值,不要将其用于任何后续查询。

如需了解详情,请参阅变更数据流查询工作流。

空闲事务

如果事务没有未完成的读取或 SQL 查询,并且在过去 10 秒内未启动,则会视为空闲。Spanner 可以取消空闲事务,以防止它们无限期地持有锁定。如果空闲事务已中止,则提交会失败并返回 ABORTED 错误。在事务内定期执行小型查询(例如 SELECT 1)可以防止其变为空闲状态。

后续步骤