/*

Copyright 2017 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.

*/

package spanner

import (

"context"

"errors"

"fmt"

"sync"

"sync/atomic"

"time"

"cloud.google.com/go/internal/trace"

sppb "cloud.google.com/go/spanner/apiv1/spannerpb"

"github.com/googleapis/gax-go/v2"

"github.com/googleapis/gax-go/v2/apierror"

"google.golang.org/api/iterator"

"google.golang.org/grpc"

"google.golang.org/grpc/codes"

"google.golang.org/grpc/metadata"

"google.golang.org/grpc/status"

"google.golang.org/protobuf/proto"

durationpb "google.golang.org/protobuf/types/known/durationpb"

)

// transactionID stores a transaction ID which uniquely identifies a transaction

// in Cloud Spanner.

type transactionID []byte

// txReadEnv manages a read-transaction environment consisting of a session

// handle and a transaction selector.

type txReadEnv interface {

// acquire returns a read-transaction environment that can be used to

// perform a transactional read.

acquire(ctx context.Context) (*sessionHandle, *sppb.TransactionSelector, error)

// getTransactionSelector returns the transaction selector based on state of the transaction it is in

getTransactionSelector() *sppb.TransactionSelector

// sets the transactionID

setTransactionID(id transactionID)

// updatePrecommitToken updates the precommit token for the transaction

updatePrecommitToken(token *sppb.MultiplexedSessionPrecommitToken)

// sets the transaction's read timestamp

setTimestamp(time.Time)

// release should be called at the end of every transactional read to deal

// with session recycling.

release(error)

}

// txReadOnly contains methods for doing transactional reads.

type txReadOnly struct {

// read-transaction environment for performing transactional read

// operations.

txReadEnv

// updateTxStateFunc is a function that updates the state of the current

// transaction based on the given error. This function is by default a no-op,

// but is overridden for read/write transactions to set the state to txAborted

// if Spanner aborts the transaction.

updateTxStateFunc func(err error) error

// Atomic. Only needed for DML statements, but used forall.

sequenceNumber int64

// sm is the session manager for allocating a session to execute the read-only

// transaction. It is set only once during initialization of the

// txReadOnly.

sm *sessionManager

// sh is the sessionHandle allocated from the session manager.

sh *sessionHandle

// qo provides options for executing a sql query.

qo QueryOptions

// ro provides options for reading rows from a database.

ro ReadOptions

// txOpts provides options for a transaction.

txOpts TransactionOptions

// commonTags for opencensus metrics

ct *commonTags

// disableRouteToLeader specifies if all the requests of type read-write and PDML

// need to be routed to the leader region.

disableRouteToLeader bool

// clientContext provides the default client context for the transaction.

clientContext *sppb.RequestOptions_ClientContext

otConfig *openTelemetryConfig

}

func (t *txReadOnly) isDefaultInlinedBegin() bool {

return false

}

func (t *txReadOnly) updateTxState(err error) error {

if t.updateTxStateFunc == nil {

return err

}

return t.updateTxStateFunc(err)

}

// BeginTransactionOption determines how a transaction is started by the client. A transaction can be started by

// inlining the BeginTransaction option with the first statement in the transaction, or by executing an explicit

// BeginTransaction RPC. Inlining the BeginTransaction with the first statement requires one less round-trip to Spanner.

// Using an explicit BeginTransaction RPC can be more efficient if you want to execute multiple queries in parallel at

// the start of the transaction, as only one statement can include a BeginTransaction option, and all other queries

// have to wait for the first query to return at least one result before proceeding.

type BeginTransactionOption int

const (

// DefaultBeginTransaction instructs the transaction to use the default for the type of transaction. The defaults are:

// * ReadWriteTransaction: InlinedBeginTransaction

// * ReadWriteStmtBasedTransaction: ExplicitBeginTransaction

// * ReadOnlyTransaction: ExplicitBeginTransaction

DefaultBeginTransaction BeginTransactionOption = iota

// InlinedBeginTransaction instructs the transaction to include the BeginTransaction with the first statement in the

// transaction. This is more efficient if the transaction does not execute any statements in parallel at the start

// of the transaction. This option is the default for ReadWriteTransaction.

InlinedBeginTransaction

// ExplicitBeginTransaction instructs the transaction to use a separate BeginTransaction RPC to start the transaction.

// This can be more efficient if the transaction executes multiple statements in parallel at the start of the

// transaction. This option is the default for ReadOnlyTransaction and ReadWriteStmtBasedTransaction.

ExplicitBeginTransaction

)

// TransactionOptions provides options for a transaction.

type TransactionOptions struct {

CommitOptions CommitOptions

// The transaction tag to use for a read/write transaction.

// This tag is automatically included with each statement and the commit

// request of a read/write transaction.

TransactionTag string

// CommitPriority is the priority to use for the Commit RPC for the

// transaction.

CommitPriority sppb.RequestOptions_Priority

// the transaction lock mode is used to specify a concurrency mode for the

// read/query operations. It works for a read/write transaction only.

ReadLockMode sppb.TransactionOptions_ReadWrite_ReadLockMode

// Controls whether to exclude recording modifications in current transaction

// from the allowed tracking change streams(with DDL option allow_txn_exclusion=true).

ExcludeTxnFromChangeStreams bool

// sets the isolation level for RW transaction

IsolationLevel sppb.TransactionOptions_IsolationLevel

// BeginTransactionOption controls whether a separate BeginTransaction RPC should be used,

// or whether the BeginTransaction operation should be inlined with the first statement

// in the transaction.

BeginTransactionOption BeginTransactionOption

// ClientContext contains client-owned context information to be passed with the transaction.

ClientContext *sppb.RequestOptions_ClientContext

}

// merge combines two TransactionOptions that the input parameter will have higher

// order of precedence.

func (to TransactionOptions) merge(opts TransactionOptions) TransactionOptions {

merged := TransactionOptions{

CommitOptions: to.CommitOptions.merge(opts.CommitOptions),

TransactionTag: to.TransactionTag,

CommitPriority: to.CommitPriority,

ExcludeTxnFromChangeStreams: to.ExcludeTxnFromChangeStreams || opts.ExcludeTxnFromChangeStreams,

IsolationLevel: to.IsolationLevel,

BeginTransactionOption: to.BeginTransactionOption,

ClientContext: mergeClientContext(to.ClientContext, opts.ClientContext),

}

if opts.TransactionTag != "" {

merged.TransactionTag = opts.TransactionTag

}

if opts.CommitPriority != sppb.RequestOptions_PRIORITY_UNSPECIFIED {

merged.CommitPriority = opts.CommitPriority

}

if opts.ReadLockMode != sppb.TransactionOptions_ReadWrite_READ_LOCK_MODE_UNSPECIFIED {

merged.ReadLockMode = opts.ReadLockMode

}

if opts.IsolationLevel != sppb.TransactionOptions_ISOLATION_LEVEL_UNSPECIFIED {

merged.IsolationLevel = opts.IsolationLevel

}

if opts.BeginTransactionOption != DefaultBeginTransaction {

merged.BeginTransactionOption = opts.BeginTransactionOption

}

return merged

}

// errSessionClosed returns error for using a recycled/destroyed session

func errSessionClosed(sh *sessionHandle) error {

return spannerErrorf(codes.FailedPrecondition,

"session is already recycled / destroyed: session_id = %q, rpc_client = %v", sh.getID(), sh.getClient())

}

// Read returns a RowIterator for reading multiple rows from the database.

func (t *txReadOnly) Read(ctx context.Context, table string, keys KeySet, columns []string) *RowIterator {

return t.ReadWithOptions(ctx, table, keys, columns, nil)

}

// ReadUsingIndex calls ReadWithOptions with ReadOptions{Index: index}.

func (t *txReadOnly) ReadUsingIndex(ctx context.Context, table, index string, keys KeySet, columns []string) (ri *RowIterator) {

return t.ReadWithOptions(ctx, table, keys, columns, &ReadOptions{Index: index})

}

// ReadOptions provides options for reading rows from a database.

type ReadOptions struct {

// The index to use for reading. If non-empty, you can only read columns

// that are part of the index key, part of the primary key, or stored in the

// index due to a STORING clause in the index definition.

Index string

// The maximum number of rows to read. A limit value less than 1 means no

// limit.

Limit int

// Priority is the RPC priority to use for the operation.

Priority sppb.RequestOptions_Priority

// The request tag to use for this request.

RequestTag string

// If this is for a partitioned read and DataBoostEnabled field is set to true, the request will be executed

// via Spanner independent compute resources. Setting this option for regular read operations has no effect.

DataBoostEnabled bool

// ReadOptions option used to set the DirectedReadOptions for all ReadRequests which indicate

// which replicas or regions should be used for running read operations.

DirectedReadOptions *sppb.DirectedReadOptions

// An option to control the order in which rows are returned from a read.

OrderBy sppb.ReadRequest_OrderBy

// A lock hint mechanism to use for this request. This setting is only applicable for

// read-write transaction as as read-only transactions do not take locks.

LockHint sppb.ReadRequest_LockHint

// ClientContext contains client-owned context information to be passed with the read request.

ClientContext *sppb.RequestOptions_ClientContext

}

// merge combines two ReadOptions that the input parameter will have higher

// order of precedence.

func (ro ReadOptions) merge(opts ReadOptions) ReadOptions {

merged := ReadOptions{

Index: ro.Index,

Limit: ro.Limit,

Priority: ro.Priority,

RequestTag: ro.RequestTag,

DataBoostEnabled: ro.DataBoostEnabled,

DirectedReadOptions: ro.DirectedReadOptions,

OrderBy: ro.OrderBy,

LockHint: ro.LockHint,

ClientContext: mergeClientContext(ro.ClientContext, opts.ClientContext),

}

if opts.Index != "" {

merged.Index = opts.Index

}

if opts.Limit > 0 {

merged.Limit = opts.Limit

}

if opts.Priority != sppb.RequestOptions_PRIORITY_UNSPECIFIED {

merged.Priority = opts.Priority

}

if opts.RequestTag != "" {

merged.RequestTag = opts.RequestTag

}

if opts.DataBoostEnabled {

merged.DataBoostEnabled = opts.DataBoostEnabled

}

if opts.DirectedReadOptions != nil {

merged.DirectedReadOptions = opts.DirectedReadOptions

}

if opts.OrderBy != sppb.ReadRequest_ORDER_BY_UNSPECIFIED {

merged.OrderBy = opts.OrderBy

}

if opts.LockHint != sppb.ReadRequest_LOCK_HINT_UNSPECIFIED {

merged.LockHint = opts.LockHint

}

return merged

}

// ReadWithOptions returns a RowIterator for reading multiple rows from the

// database. Pass a ReadOptions to modify the read operation.

func (t *txReadOnly) ReadWithOptions(ctx context.Context, table string, keys KeySet, columns []string, opts *ReadOptions) (ri *RowIterator) {

ctx, _ = startSpan(ctx, "Read", t.otConfig.commonTraceStartOptions...)

defer func() { endSpan(ctx, ri.err) }()

var (

sh *sessionHandle

ts *sppb.TransactionSelector

err error

)

kset, err := keys.keySetProto()

if err != nil {

return &RowIterator{

meterTracerFactory: t.sm.sc.metricsTracerFactory,

err: err}

}

if sh, ts, err = t.acquire(ctx); err != nil {

return &RowIterator{

meterTracerFactory: t.sm.sc.metricsTracerFactory,

err: err}

}

// Cloud Spanner will return "Session not found" on bad sessions.

client := sh.getClient()

if client == nil {

// Might happen if transaction is closed in the middle of a API call.

return &RowIterator{

meterTracerFactory: t.sm.sc.metricsTracerFactory,

err: errSessionClosed(sh)}

}

index := t.ro.Index

limit := t.ro.Limit

prio := t.ro.Priority

requestTag := t.ro.RequestTag

dataBoostEnabled := t.ro.DataBoostEnabled

directedReadOptions := t.ro.DirectedReadOptions

orderBy := t.ro.OrderBy

lockHint := t.ro.LockHint

clientContext := t.ro.ClientContext

if opts != nil {

index = opts.Index

if opts.Limit > 0 {

limit = opts.Limit

}

prio = opts.Priority

requestTag = opts.RequestTag

if opts.DataBoostEnabled {

dataBoostEnabled = opts.DataBoostEnabled

}

if opts.DirectedReadOptions != nil {

directedReadOptions = opts.DirectedReadOptions

}

if opts.OrderBy != sppb.ReadRequest_ORDER_BY_UNSPECIFIED {

orderBy = opts.OrderBy

}

if opts.LockHint != sppb.ReadRequest_LOCK_HINT_UNSPECIFIED {

lockHint = opts.LockHint

}

clientContext = mergeClientContext(clientContext, opts.ClientContext)

}

var setTransactionID func(transactionID)

if _, ok := ts.Selector.(*sppb.TransactionSelector_Begin); ok {

setTransactionID = t.setTransactionID

} else {

setTransactionID = nil

}

return streamWithTransactionCallbacks(

contextWithOutgoingMetadata(ctx, sh.getMetadata(), t.disableRouteToLeader),

sh.session.logger,

t.sm.sc.metricsTracerFactory,

func(ctx context.Context, resumeToken []byte, opts ...gax.CallOption) (streamingReceiver, error) {

req := &sppb.ReadRequest{

Session: t.sh.getID(),

Transaction: t.getTransactionSelector(),

Table: table,

Index: index,

Columns: columns,

KeySet: kset,

ResumeToken: resumeToken,

Limit: int64(limit),

RequestOptions: createRequestOptions(prio, requestTag, t.txOpts.TransactionTag, mergeClientContext(t.clientContext, clientContext)),

DataBoostEnabled: dataBoostEnabled,

DirectedReadOptions: directedReadOptions,

OrderBy: orderBy,

LockHint: lockHint,

}

client, err := client.StreamingRead(ctx, req, opts...)

if err != nil {

if _, ok := t.getTransactionSelector().GetSelector().(*sppb.TransactionSelector_Begin); ok {

t.setTransactionID(nil)

return client, t.updateTxState(errInlineBeginTransactionFailed(err))

}

return client, t.updateTxState(err)

}

md, err := client.Header()

if getGFELatencyMetricsFlag() && md != nil && t.ct != nil {

if err := createContextAndCaptureGFELatencyMetrics(ctx, t.ct, md, "ReadWithOptions"); err != nil {

trace.TracePrintf(ctx, nil, "Error in recording GFE Latency. Try disabling and rerunning. Error: %v", err)

}

}

if metricErr := recordGFELatencyMetricsOT(ctx, md, "ReadWithOptions", t.otConfig); metricErr != nil {

trace.TracePrintf(ctx, nil, "Error in recording GFE Latency through OpenTelemetry. Error: %v", metricErr)

}

return client, err

},

setTransactionID,

func(err error) error {

return t.updateTxState(err)

},

t.updatePrecommitToken,

t.setTimestamp,

t.release,

asGRPCSpannerClient(client),

)

}

// errRowNotFound returns error for not being able to read the row identified by

// key.

func errRowNotFound(table string, key Key) error {

err := spannerErrorf(codes.NotFound, "row not found(Table: %v, PrimaryKey: %v)", table, key)

err.(*Error).err = ErrRowNotFound

return err

}

// errRowNotFoundByIndex returns error for not being able to read the row by index.

func errRowNotFoundByIndex(table string, key Key, index string) error {

err := spannerErrorf(codes.NotFound, "row not found(Table: %v, IndexKey: %v, Index: %v)", table, key, index)

err.(*Error).err = ErrRowNotFound

return err

}

// errMultipleRowsFound returns error for receiving more than one row when reading a single row using an index.

func errMultipleRowsFound(table string, key Key, index string) error {

return spannerErrorf(codes.FailedPrecondition, "more than one row found by index(Table: %v, IndexKey: %v, Index: %v)", table, key, index)

}

func errTransactionNoLongerActive() error {

return spannerError(codes.FailedPrecondition, "the transaction that was used to execute this statement is no longer active")

}

const errInlineBeginTransactionFailedMsg = "failed inline begin transaction"

// errInlineBeginTransactionFailed creates an error that indicates that the first statement in the

// transaction included an inline-begin-transaction option, and that the statement failed. This

// internal error is an indication to the client library that the transaction should be retried

// with an explicit BeginTransaction RPC.

//

// The error that is passed in to this function is wrapped in the error that is returned. It is

// ignored if it is nil.

func errInlineBeginTransactionFailed(err error) error {

// Create an API error that can be wrapped in the error that we return.

// This API error is used by the client library to determine the error code.

apiErr, _ := apierror.FromError(status.Error(codes.Internal, errInlineBeginTransactionFailedMsg))

var wrapped error

if err == nil {

// Only wrap the API error if there is no user error.

wrapped = apiErr

} else {

// Create a combination of both the API error and the given user error and wrap

// that in the internal error that will be returned.

wrapped = fmt.Errorf("%s: %w (%w)", errInlineBeginTransactionFailedMsg, apiErr, err)

}

return &Error{

Code: codes.Internal,

err: wrapped,

Desc: errInlineBeginTransactionFailedMsg,

}

}

// ReadRow reads a single row from the database.

//

// If no row is present with the given key, then ReadRow returns an error(spanner.ErrRowNotFound) where

// spanner.ErrCode(err) is codes.NotFound.

//

// To check if the error is spanner.ErrRowNotFound:

//

// if errors.Is(err, spanner.ErrRowNotFound) {

// ...

// }

func (t *txReadOnly) ReadRow(ctx context.Context, table string, key Key, columns []string) (*Row, error) {

return t.ReadRowWithOptions(ctx, table, key, columns, nil)

}

// ReadRowWithOptions reads a single row from the database. Pass a ReadOptions to modify the read operation.

//

// If no row is present with the given key, then ReadRowWithOptions returns an error where

// spanner.ErrCode(err) is codes.NotFound.

//

// To check if the error is spanner.ErrRowNotFound:

//

// if errors.Is(err, spanner.ErrRowNotFound) {

// ...

// }

func (t *txReadOnly) ReadRowWithOptions(ctx context.Context, table string, key Key, columns []string, opts *ReadOptions) (*Row, error) {

iter := t.ReadWithOptions(ctx, table, key, columns, opts)

defer iter.Stop()

row, err := iter.Next()

switch err {

case iterator.Done:

return nil, errRowNotFound(table, key)

case nil:

return row, nil

default:

return nil, err

}

}

// ReadRowUsingIndex reads a single row from the database using an index.

//

// If no row is present with the given index, then ReadRowUsingIndex returns an

// error(spanner.ErrRowNotFound) where spanner.ErrCode(err) is codes.NotFound.

//

// To check if the error is spanner.ErrRowNotFound:

//

// if errors.Is(err, spanner.ErrRowNotFound) {

// ...

// }

//

// If more than one row received with the given index, then ReadRowUsingIndex

// returns an error where spanner.ErrCode(err) is codes.FailedPrecondition.

func (t *txReadOnly) ReadRowUsingIndex(ctx context.Context, table string, index string, key Key, columns []string) (*Row, error) {

iter := t.ReadUsingIndex(ctx, table, index, key, columns)

defer iter.Stop()

row, err := iter.Next()

switch err {

case iterator.Done:

return nil, errRowNotFoundByIndex(table, key, index)

case nil:

// If more than one row found, return an error.

_, err := iter.Next()

switch err {

case iterator.Done:

return row, nil

case nil:

return nil, errMultipleRowsFound(table, key, index)

default:

return nil, err

}

default:

return nil, err

}

}

// QueryOptions provides options for executing a sql query or update statement.

type QueryOptions struct {

Mode *sppb.ExecuteSqlRequest_QueryMode

Options *sppb.ExecuteSqlRequest_QueryOptions

// Priority is the RPC priority to use for the query/update.

Priority sppb.RequestOptions_Priority

// The request tag to use for this request.

RequestTag string

// If this is for a partitioned query and DataBoostEnabled field is set to true, the request will be executed

// via Spanner independent compute resources. Setting this option for regular query operations has no effect.

DataBoostEnabled bool

// QueryOptions option used to set the DirectedReadOptions for all ExecuteSqlRequests which indicate

// which replicas or regions should be used for executing queries.

DirectedReadOptions *sppb.DirectedReadOptions

// Controls whether to exclude recording modifications in current partitioned update operation

// from the allowed tracking change streams(with DDL option allow_txn_exclusion=true). Setting

// this value for any sql/dml requests other than partitioned update will receive an error.

ExcludeTxnFromChangeStreams bool

// LastStatement indicates whether this statement is the last statement in this transaction.

// If set to true, this option marks the end of the transaction. The transaction should be

// committed or rolled back after this statement executes, and attempts to execute any other requests

// against this transaction (including reads and queries) will be rejected. Mixing mutations with

// statements that are marked as the last statement is not allowed.

//

// For DML statements, setting this option may cause some error reporting to be deferred until

// commit time (e.g. validation of unique constraints). Given this, successful execution of a DML

// statement should not be assumed until the transaction commits.

LastStatement bool

// ClientContext contains client-owned context information to be passed with the query.

ClientContext *sppb.RequestOptions_ClientContext

}

// merge combines two QueryOptions that the input parameter will have higher

// order of precedence.

func (qo QueryOptions) merge(opts QueryOptions) QueryOptions {

merged := QueryOptions{

Mode: qo.Mode,

Options: &sppb.ExecuteSqlRequest_QueryOptions{},

RequestTag: qo.RequestTag,

Priority: qo.Priority,

DataBoostEnabled: qo.DataBoostEnabled,

DirectedReadOptions: qo.DirectedReadOptions,

ExcludeTxnFromChangeStreams: qo.ExcludeTxnFromChangeStreams || opts.ExcludeTxnFromChangeStreams,

LastStatement: qo.LastStatement || opts.LastStatement,

ClientContext: mergeClientContext(qo.ClientContext, opts.ClientContext),

}

if opts.Mode != nil {

merged.Mode = opts.Mode

}

if opts.RequestTag != "" {

merged.RequestTag = opts.RequestTag

}

if opts.Priority != sppb.RequestOptions_PRIORITY_UNSPECIFIED {

merged.Priority = opts.Priority

}

if opts.DataBoostEnabled {

merged.DataBoostEnabled = opts.DataBoostEnabled

}

if opts.DirectedReadOptions != nil {

merged.DirectedReadOptions = opts.DirectedReadOptions

}

proto.Merge(merged.Options, qo.Options)

proto.Merge(merged.Options, opts.Options)

return merged

}

func mergeClientContext(defaultCtx, specificCtx *sppb.RequestOptions_ClientContext) *sppb.RequestOptions_ClientContext {

if defaultCtx == nil {

return specificCtx

}

if specificCtx == nil {

return defaultCtx

}

merged := proto.Clone(defaultCtx).(*sppb.RequestOptions_ClientContext)

proto.Merge(merged, specificCtx)

return merged

}

func createRequestOptions(prio sppb.RequestOptions_Priority, requestTag, transactionTag string, clientContext *sppb.RequestOptions_ClientContext) (ro *sppb.RequestOptions) {

ro = &sppb.RequestOptions{}

if prio != sppb.RequestOptions_PRIORITY_UNSPECIFIED {

ro.Priority = prio

}

if requestTag != "" {

ro.RequestTag = requestTag

}

if transactionTag != "" {

ro.TransactionTag = transactionTag

}

if clientContext != nil {

ro.ClientContext = clientContext

}

return ro

}

// 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.

// Use QueryWithStats to get rows along with the plan and statistics. Use

// AnalyzeQuery to get just the plan.

func (t *txReadOnly) Query(ctx context.Context, statement Statement) *RowIterator {

mode := sppb.ExecuteSqlRequest_NORMAL

return t.query(ctx, statement, QueryOptions{

Mode: &mode,

Options: t.qo.Options,

Priority: t.qo.Priority,

DirectedReadOptions: t.qo.DirectedReadOptions,

})

}

// QueryWithOptions executes a SQL statment against the database. It returns

// a RowIterator for retrieving the resulting rows. The sql query execution

// will be optimized based on the given query options.

func (t *txReadOnly) QueryWithOptions(ctx context.Context, statement Statement, opts QueryOptions) *RowIterator {

return t.query(ctx, statement, t.qo.merge(opts))

}

// QueryWithStats executes a SQL statement against the database. It returns

// a RowIterator for retrieving the resulting rows. The RowIterator will also

// be populated with a query plan and execution statistics.

func (t *txReadOnly) QueryWithStats(ctx context.Context, statement Statement) *RowIterator {

mode := sppb.ExecuteSqlRequest_PROFILE

return t.query(ctx, statement, QueryOptions{

Mode: &mode,

Options: t.qo.Options,

Priority: t.qo.Priority,

DirectedReadOptions: t.qo.DirectedReadOptions,

})

}

// AnalyzeQuery returns the query plan for statement.

func (t *txReadOnly) AnalyzeQuery(ctx context.Context, statement Statement) (*sppb.QueryPlan, error) {

mode := sppb.ExecuteSqlRequest_PLAN

iter := t.query(ctx, statement, QueryOptions{

Mode: &mode,

Options: t.qo.Options,

Priority: t.qo.Priority,

DirectedReadOptions: t.qo.DirectedReadOptions,

})

defer iter.Stop()

for {

_, err := iter.Next()

if err == iterator.Done {

break

}

if err != nil {

return nil, err

}

}

if iter.QueryPlan == nil {

return nil, spannerErrorf(codes.Internal, "query plan unavailable")

}

return iter.QueryPlan, nil

}

func (t *txReadOnly) query(ctx context.Context, statement Statement, options QueryOptions) (ri *RowIterator) {

ctx, _ = startSpan(ctx, "Query", t.otConfig.commonTraceStartOptions...)

defer func() { endSpan(ctx, ri.err) }()

req, sh, err := t.prepareExecuteSQL(ctx, statement, options)

if err != nil {

return &RowIterator{

meterTracerFactory: t.sm.sc.metricsTracerFactory,

err: err,

}

}

var setTransactionID func(transactionID)

if _, ok := req.Transaction.GetSelector().(*sppb.TransactionSelector_Begin); ok {

setTransactionID = t.setTransactionID

} else {

setTransactionID = nil

}

client := sh.getClient()

return streamWithTransactionCallbacks(

contextWithOutgoingMetadata(ctx, sh.getMetadata(), t.disableRouteToLeader),

sh.session.logger,

t.sm.sc.metricsTracerFactory,

func(ctx context.Context, resumeToken []byte, opts ...gax.CallOption) (streamingReceiver, error) {

// The session handle is removed from the transaction when the transaction is committed or rolled back.

// This ensures that we return a reasonable error instead of panic if the application tries to use the

// stream after the transaction has finished.

if t.sh == nil {

return nil, errTransactionNoLongerActive()

}

req.ResumeToken = resumeToken

req.Session = t.sh.getID()

req.Transaction = t.getTransactionSelector()

client, err := client.ExecuteStreamingSql(ctx, req, opts...)

if err != nil {

if _, ok := req.Transaction.GetSelector().(*sppb.TransactionSelector_Begin); ok {

t.setTransactionID(nil)

return client, t.updateTxState(errInlineBeginTransactionFailed(err))

}

return client, t.updateTxState(err)

}

md, err := client.Header()

if getGFELatencyMetricsFlag() && md != nil && t.ct != nil {

if err := createContextAndCaptureGFELatencyMetrics(ctx, t.ct, md, "query"); err != nil {

trace.TracePrintf(ctx, nil, "Error in recording GFE Latency. Try disabling and rerunning. Error: %v", err)

}

}

if metricErr := recordGFELatencyMetricsOT(ctx, md, "query", t.otConfig); metricErr != nil {

trace.TracePrintf(ctx, nil, "Error in recording GFE Latency through OpenTelemetry. Error: %v", metricErr)

}

return client, err

},

setTransactionID,

func(err error) error {

return t.updateTxState(err)

},

t.updatePrecommitToken,

t.setTimestamp,

t.release,

asGRPCSpannerClient(client))

}

func (t *txReadOnly) prepareExecuteSQL(ctx context.Context, stmt Statement, options QueryOptions) (*sppb.ExecuteSqlRequest, *sessionHandle, error) {

sh, ts, err := t.acquire(ctx)

if err != nil {

return nil, nil, err

}

// Cloud Spanner will return "Session not found" on bad sessions.

sid := sh.getID()

if sid == "" {

// Might happen if transaction is closed in the middle of a API call.

return nil, nil, errSessionClosed(sh)

}

params, paramTypes, err := stmt.convertParams()

if err != nil {

return nil, nil, err

}

mode := sppb.ExecuteSqlRequest_NORMAL

if options.Mode != nil {

mode = *options.Mode

}

req := &sppb.ExecuteSqlRequest{

Session: sid,

Transaction: ts,

Sql: stmt.SQL,

QueryMode: mode,

Seqno: atomic.AddInt64(&t.sequenceNumber, 1),

Params: params,

ParamTypes: paramTypes,

QueryOptions: options.Options,

RequestOptions: createRequestOptions(options.Priority, options.RequestTag, t.txOpts.TransactionTag, mergeClientContext(t.clientContext, options.ClientContext)),

DataBoostEnabled: options.DataBoostEnabled,

DirectedReadOptions: options.DirectedReadOptions,

LastStatement: options.LastStatement,

}

return req, sh, nil

}

// txState is the status of a transaction.

type txState int

const (

// transaction is new, waiting to be initialized..

txNew txState = iota

// transaction is being initialized.

txInit

// transaction is active and can perform read/write.

txActive

// transaction is closed, cannot be used anymore.

txClosed

// transaction was aborted by Spanner and should be retried.

txAborted

)

// errRtsUnavailable returns error for read transaction's read timestamp being

// unavailable.

func errRtsUnavailable() error {

return spannerErrorf(codes.Internal, "read timestamp is unavailable")

}

// errTxClosed returns error for using a closed transaction.

func errTxClosed() error {

return spannerErrorf(codes.InvalidArgument, "cannot use a closed transaction")

}

// errUnexpectedTxState returns error for transaction enters an unexpected state.

func errUnexpectedTxState(ts txState) error {

return spannerErrorf(codes.FailedPrecondition, "unexpected transaction state: %v", ts)

}

// errExcludeRequestLevelDmlFromChangeStreams returns error for passing

// QueryOptions.ExcludeTxnFromChangeStreams to request-level DML functions. This

// options should only be used for partitioned update.

func errExcludeRequestLevelDmlFromChangeStreams() error {

return spannerErrorf(codes.InvalidArgument, "cannot set exclude transaction from change streams for a request-level DML statement.")

}

// ReadOnlyTransaction provides a snapshot transaction with guaranteed

// consistency across reads, but does not allow writes. Read-only transactions

// can be configured to read at timestamps in the past.

//

// Read-only transactions do not take locks. Instead, they work by choosing a

// Cloud Spanner timestamp, then executing all reads at that timestamp. Since

// they do not acquire locks, they do not block concurrent read-write

// transactions.

//

// Unlike locking read-write transactions, read-only transactions never abort.

// They can fail if the chosen read timestamp is garbage collected; however, the

// default garbage collection policy is generous enough that most applications

// do not need to worry about this in practice. See the documentation of

// TimestampBound for more details.

//

// A ReadOnlyTransaction consumes resources on the server until Close is called.

type ReadOnlyTransaction struct {

// mu protects concurrent access to the internal states of ReadOnlyTransaction.

mu sync.Mutex

// txReadOnly contains methods for performing transactional reads.

txReadOnly

// singleUse indicates that the transaction can be used for only one read.

singleUse bool

// tx is the transaction ID in Cloud Spanner that uniquely identifies the

// ReadOnlyTransaction.

tx transactionID

// txReadyOrClosed is for broadcasting that transaction ID has been returned

// by Cloud Spanner or that transaction is closed.

txReadyOrClosed chan struct{}

// state is the current transaction status of the ReadOnly transaction.

state txState

// rts is the read timestamp returned by transactional reads.

rts time.Time

// tb is the read staleness bound specification for transactional reads.

tb TimestampBound

// beginTransactionOption determines whether the transaction should use inlined-begin

// or an explicit BeginTransaction RPC.

beginTransactionOption BeginTransactionOption

// isLongRunningTransaction indicates whether the transaction is long-running or not.

isLongRunningTransaction bool

}

// errTxInitTimeout returns error for timeout in waiting for initialization of

// the transaction.

func errTxInitTimeout() error {

return spannerErrorf(codes.Canceled, "timeout/context canceled in waiting for transaction's initialization")

}

// getTimestampBound returns the read staleness bound specified for the

// ReadOnlyTransaction.

func (t *ReadOnlyTransaction) getTimestampBound() TimestampBound {

t.mu.Lock()

defer t.mu.Unlock()

return t.tb

}

// begin starts a snapshot read-only Transaction on Cloud Spanner.

func (t *ReadOnlyTransaction) begin(ctx context.Context) error {

var (

locked bool

tx transactionID

rts time.Time

sh *sessionHandle

err error

res *sppb.Transaction

)

defer func() {

if !locked {

t.mu.Lock()

// Not necessary, just to make it clear that t.mu is being held when

// locked == true.

locked = true

}

if t.state != txClosed {

// Signal other initialization routines.

close(t.txReadyOrClosed)

t.txReadyOrClosed = make(chan struct{})

}

t.mu.Unlock()

if err != nil && sh != nil {

// Got a valid session handle, but failed to initialize transaction

// on Cloud Spanner.

sh.recycle()

}

}()

sh, err = t.sm.takeMultiplexed(ctx)

if err != nil {

return err

}

var md metadata.MD

res, err = sh.getClient().BeginTransaction(contextWithOutgoingMetadata(ctx, sh.getMetadata(), t.disableRouteToLeader), &sppb.BeginTransactionRequest{

Session: sh.getID(),

Options: &sppb.TransactionOptions{

Mode: &sppb.TransactionOptions_ReadOnly_{

ReadOnly: buildTransactionOptionsReadOnly(t.getTimestampBound(), true),

},

},

RequestOptions: createRequestOptions(sppb.RequestOptions_PRIORITY_UNSPECIFIED, "", "", t.clientContext),

}, gax.WithGRPCOptions(grpc.Header(&md)))

if getGFELatencyMetricsFlag() && md != nil && t.ct != nil {

if err := createContextAndCaptureGFELatencyMetrics(ctx, t.ct, md, "begin_BeginTransaction"); err != nil {

trace.TracePrintf(ctx, nil, "Error in recording GFE Latency. Try disabling and rerunning. Error: %v", err)

}

}

if metricErr := recordGFELatencyMetricsOT(ctx, md, "begin_BeginTransaction", t.otConfig); metricErr != nil {

trace.TracePrintf(ctx, nil, "Error in recording GFE Latency through OpenTelemetry. Error: %v", metricErr)

}

if err == nil {

tx = res.Id

if res.ReadTimestamp != nil {

rts = time.Unix(res.ReadTimestamp.Seconds, int64(res.ReadTimestamp.Nanos))

}

} else {

err = ToSpannerError(err)

}

t.mu.Lock()

// defer function will be executed with t.mu being held.

locked = true

// During the execution of t.begin(), t.Close() was invoked.

if t.state == txClosed {

return errSessionClosed(sh)

}

// If begin() fails, this allows other queries to take over the

// initialization.

t.tx = nil

if err == nil {

t.tx = tx

t.rts = rts

t.sh = sh

// State transite to txActive.

t.state = txActive

}

return err

}

// no-op for ReadOnlyTransaction.

func (t *ReadOnlyTransaction) updatePrecommitToken(token *sppb.MultiplexedSessionPrecommitToken) {

return

}

// acquire implements txReadEnv.acquire.

func (t *ReadOnlyTransaction) acquire(ctx context.Context) (*sessionHandle, *sppb.TransactionSelector, error) {

if err := checkNestedTxn(ctx); err != nil {

return nil, nil, err

}

if t.singleUse {

return t.acquireSingleUse(ctx)

}

return t.acquireMultiUse(ctx)

}

func (t *ReadOnlyTransaction) acquireSingleUse(ctx context.Context) (*sessionHandle, *sppb.TransactionSelector, error) {

t.mu.Lock()

defer t.mu.Unlock()

switch t.state {