chore: delay transaction activation until actual use

Delay the actual transaction activation until the first actual usage of the transaction.
That is; the first time that a statement is being sent to Spanner. This allows the
application to amend the transaction options after calling BeginTx or executing
`BEGIN TRANSACTION`. The transaction options can be amended by executing a statement
like `SET TRANSACTION READ ONLY`.

Author: olavloite

aborted_transactions_test.go

@@ -41,6 +41,9 @@ func TestCommitAborted(t *testing.T) {
 	if err != nil {
 		t.Fatalf("begin failed: %v", err)
 	}
+	if _, err := tx.ExecContext(ctx, testutil.UpdateBarSetFoo); err != nil {
+		t.Fatal(err)
+	}
 	server.TestSpanner.PutExecutionTime(testutil.MethodCommitTransaction, testutil.SimulatedExecutionTime{
 		Errors: []error{status.Error(codes.Aborted, "Aborted")},
 	})
@@ -51,7 +54,7 @@ func TestCommitAborted(t *testing.T) {
 	reqs := server.TestSpanner.DrainRequestsFromServer()
 	commitReqs := testutil.RequestsOfType(reqs, reflect.TypeOf(&sppb.CommitRequest{}))
 	if g, w := len(commitReqs), 2; g != w {
-		t.Fatalf("commit request count mismatch\nGot: %v\nWant: %v", g, w)
+		t.Fatalf("commit request count mismatch\n Got: %v\nWant: %v", g, w)
 	}
 
 	// Verify that the db is still usable.
@@ -117,6 +120,9 @@ func TestCommitAbortedWithInternalRetriesDisabled(t *testing.T) {
 	if err != nil {
 		t.Fatalf("begin failed: %v", err)
 	}
+	if _, err := tx.ExecContext(ctx, testutil.UpdateBarSetFoo); err != nil {
+		t.Fatal(err)
+	}
 	server.TestSpanner.PutExecutionTime(testutil.MethodCommitTransaction, testutil.SimulatedExecutionTime{
 		Errors: []error{status.Error(codes.Aborted, "Aborted")},
 	})

client_side_statement_test.go

@@ -65,7 +65,7 @@ func TestStatementExecutor_StartBatchDdl(t *testing.T) {
 	}
 
 	// Starting a DDL batch while the connection is in a transaction is not allowed.
-	c.tx = &readWriteTransaction{}
+	c.tx = &delegatingTransaction{conn: c, ctx: ctx}
 	if _, err := c.ExecContext(ctx, "start batch ddl", []driver.NamedValue{}); spanner.ErrCode(err) != codes.FailedPrecondition {
 		t.Fatalf("error mismatch for starting a DDL batch while in a transaction\nGot: %v\nWant: %v", spanner.ErrCode(err), codes.FailedPrecondition)
 	}
@@ -102,13 +102,13 @@ func TestStatementExecutor_StartBatchDml(t *testing.T) {
 	}
 
 	// Starting a DML batch while the connection is in a read-only transaction is not allowed.
-	c.tx = &readOnlyTransaction{logger: noopLogger}
+	c.tx = &delegatingTransaction{conn: c, contextTransaction: &readOnlyTransaction{logger: noopLogger}}
 	if _, err := c.ExecContext(ctx, "start batch dml", []driver.NamedValue{}); spanner.ErrCode(err) != codes.FailedPrecondition {
 		t.Fatalf("error mismatch for starting a DML batch while in a read-only transaction\nGot: %v\nWant: %v", spanner.ErrCode(err), codes.FailedPrecondition)
 	}
 
 	// Starting a DML batch while the connection is in a read/write transaction is allowed.
-	c.tx = &readWriteTransaction{logger: noopLogger}
+	c.tx = &delegatingTransaction{conn: c, contextTransaction: &readWriteTransaction{logger: noopLogger}}
 	if _, err := c.ExecContext(ctx, "start batch dml", []driver.NamedValue{}); err != nil {
 		t.Fatalf("could not start a DML batch while in a read/write transaction: %v", err)
 	}

conn.go

@@ -231,6 +231,8 @@ type SpannerConn interface {
 	// returned.
 	resetTransactionForRetry(ctx context.Context, errDuringCommit bool) error
 
+	withTransactionCloseFunc(close func())
+
 	// withTempTransactionOptions sets the TransactionOptions that should be used
 	// for the next read/write transaction. This method should only be called
 	// directly before starting a new read/write transaction.
@@ -257,8 +259,8 @@ type conn struct {
 	adminClient   *adminapi.DatabaseAdminClient
 	connId        string
 	logger        *slog.Logger
-	tx            contextTransaction
-	prevTx        contextTransaction
+	tx            *delegatingTransaction
+	prevTx        *delegatingTransaction
 	resetForRetry bool
 	database      string
 
@@ -531,7 +533,7 @@ func (c *conn) InDDLBatch() bool {
 }
 
 func (c *conn) InDMLBatch() bool {
-	return (c.batch != nil && c.batch.tp == parser.BatchTypeDml) || (c.inReadWriteTransaction() && c.tx.(*readWriteTransaction).batch != nil)
+	return (c.batch != nil && c.batch.tp == parser.BatchTypeDml) || (c.inTransaction() && c.tx.IsInBatch())
 }
 
 func (c *conn) GetBatchedStatements() []spanner.Statement {
@@ -567,9 +569,6 @@ func (c *conn) startBatchDML(automatic bool) (driver.Result, error) {
 	if c.batch != nil {
 		return nil, spanner.ToSpannerError(status.Errorf(codes.FailedPrecondition, "This connection already has an active batch."))
 	}
-	if c.inReadOnlyTransaction() {
-		return nil, spanner.ToSpannerError(status.Errorf(codes.FailedPrecondition, "This connection has an active read-only transaction. Read-only transactions cannot execute DML batches."))
-	}
 	c.logger.Debug("starting dml batch outside transaction")
 	c.batch = &batch{tp: parser.BatchTypeDml, options: execOptions}
 	return driver.ResultNoRows, nil
@@ -655,8 +654,8 @@ func (c *conn) execBatchDML(ctx context.Context, statements []spanner.Statement,
 
 	var affected []int64
 	var err error
-	if c.inTransaction() {
-		tx, ok := c.tx.(*readWriteTransaction)
+	if c.inTransaction() && c.tx.contextTransaction != nil {
+		tx, ok := c.tx.contextTransaction.(*readWriteTransaction)
 		if !ok {
 			return nil, status.Errorf(codes.FailedPrecondition, "connection is in a transaction that is not a read/write transaction")
 		}
@@ -944,7 +943,7 @@ func (c *conn) execContext(ctx context.Context, query string, execOptions *ExecO
 	}
 
 	// Start an automatic DML batch.
-	if c.AutoBatchDml() && !c.inBatch() && c.inReadWriteTransaction() {
+	if c.AutoBatchDml() && !c.inBatch() && c.inTransaction() && statementInfo.StatementType == parser.StatementTypeDml {
 		if _, err := c.startBatchDML( /* automatic = */ true); err != nil {
 			return nil, err
 		}
@@ -1041,14 +1040,14 @@ func (c *conn) resetTransactionForRetry(ctx context.Context, errDuringCommit boo
 	return c.tx.resetForRetry(ctx)
 }
 
+func (c *conn) withTransactionCloseFunc(close func()) {
+	c.tempTransactionCloseFunc = close
+}
+
 func (c *conn) withTempTransactionOptions(options *ReadWriteTransactionOptions) {
 	if options == nil {
 		return
 	}
-	c.tempTransactionCloseFunc = options.close
-	// Start a transaction for the connection state, so we can set the transaction options
-	// as local options in the current transaction.
-	_ = c.state.Begin()
 	if options.DisableInternalRetries {
 		_ = propertyRetryAbortsInternally.SetLocalValue(c.state, !options.DisableInternalRetries)
 	}
@@ -1102,10 +1101,6 @@ func (c *conn) withTempReadOnlyTransactionOptions(options *ReadOnlyTransactionOp
 	if options == nil {
 		return
 	}
-	c.tempTransactionCloseFunc = options.close
-	// Start a transaction for the connection state, so we can set the transaction options
-	// as local options in the current transaction.
-	_ = c.state.Begin()
 	if options.BeginTransactionOption != spanner.DefaultBeginTransaction {
 		_ = propertyBeginTransactionOption.SetLocalValue(c.state, options.BeginTransactionOption)
 	}
@@ -1122,10 +1117,6 @@ func (c *conn) withTempBatchReadOnlyTransactionOptions(options *BatchReadOnlyTra
 	if options == nil {
 		return
 	}
-	c.tempTransactionCloseFunc = options.close
-	// Start a transaction for the connection state, so we can set the transaction options
-	// as local options in the current transaction.
-	_ = c.state.Begin()
 	if options.TimestampBound.String() != "(strong)" {
 		_ = propertyReadOnlyStaleness.SetLocalValue(c.state, options.TimestampBound)
 	}
@@ -1139,9 +1130,9 @@ func (c *conn) getBatchReadOnlyTransactionOptions() BatchReadOnlyTransactionOpti
 // It is exported for internal reasons, and may receive breaking changes without prior notice.
 //
 // BeginReadOnlyTransaction starts a new read-only transaction on this connection.
-func (c *conn) BeginReadOnlyTransaction(ctx context.Context, options *ReadOnlyTransactionOptions) (driver.Tx, error) {
+func (c *conn) BeginReadOnlyTransaction(ctx context.Context, options *ReadOnlyTransactionOptions, close func()) (driver.Tx, error) {
+	tx, err := c.beginTx(ctx, driver.TxOptions{ReadOnly: true}, close)
 	c.withTempReadOnlyTransactionOptions(options)
-	tx, err := c.BeginTx(ctx, driver.TxOptions{ReadOnly: true})
 	if err != nil {
 		return nil, err
 	}
@@ -1152,9 +1143,9 @@ func (c *conn) BeginReadOnlyTransaction(ctx context.Context, options *ReadOnlyTr
 // It is exported for internal reasons, and may receive breaking changes without prior notice.
 //
 // BeginReadWriteTransaction starts a new read/write transaction on this connection.
-func (c *conn) BeginReadWriteTransaction(ctx context.Context, options *ReadWriteTransactionOptions) (driver.Tx, error) {
+func (c *conn) BeginReadWriteTransaction(ctx context.Context, options *ReadWriteTransactionOptions, close func()) (driver.Tx, error) {
+	tx, err := c.beginTx(ctx, driver.TxOptions{}, close)
 	c.withTempTransactionOptions(options)
-	tx, err := c.BeginTx(ctx, driver.TxOptions{})
 	if err != nil {
 		return nil, err
 	}
@@ -1177,19 +1168,6 @@ func (c *conn) beginTx(ctx context.Context, driverOpts driver.TxOptions, closeFu
 		c.resetForRetry = false
 		return c.tx, nil
 	}
-	// Also start a transaction on the ConnectionState if the BeginTx call was successful.
-	defer func() {
-		if c.tx != nil {
-			_ = c.state.Begin()
-		} else {
-			// Rollback in case the connection state transaction was started before this function
-			// was called, for example if the caller set temporary transaction options.
-			_ = c.state.Rollback()
-		}
-	}()
-
-	readOnlyTxOpts := c.getReadOnlyTransactionOptions()
-	batchReadOnlyTxOpts := c.getBatchReadOnlyTransactionOptions()
 	if c.inTransaction() {
 		return nil, spanner.ToSpannerError(status.Errorf(codes.FailedPrecondition, "already in a transaction"))
 	}
@@ -1227,94 +1205,105 @@ func (c *conn) beginTx(ctx context.Context, driverOpts driver.TxOptions, closeFu
 	if closeFunc == nil {
 		closeFunc = func() {}
 	}
+	if err := c.state.Begin(); err != nil {
+		return nil, err
+	}
+	c.clearCommitResponse()
 
+	if isolationLevelFromTxOpts != spannerpb.TransactionOptions_ISOLATION_LEVEL_UNSPECIFIED {
+		_ = propertyIsolationLevel.SetLocalValue(c.state, sql.IsolationLevel(driverOpts.Isolation))
+	}
+	// TODO: Figure out how to distinguish between 'use the default' and 'use read/write'.
 	if driverOpts.ReadOnly {
+		_ = propertyTransactionReadOnly.SetLocalValue(c.state, true)
+	}
+	if batchReadOnly {
+		_ = propertyTransactionBatchReadOnly.SetLocalValue(c.state, true)
+	}
+	if disableRetryAborts {
+		_ = propertyRetryAbortsInternally.SetLocalValue(c.state, false)
+	}
+
+	c.tx = &delegatingTransaction{
+		conn: c,
+		ctx:  ctx,
+		close: func(result txResult) {
+			closeFunc()
+			if result == txResultCommit {
+				_ = c.state.Commit()
+			} else {
+				_ = c.state.Rollback()
+			}
+			c.tx = nil
+		},
+	}
+	return c.tx, nil
+}
+
+func (c *conn) activateTransaction() (contextTransaction, error) {
+	closeFunc := c.tx.close
+	if propertyTransactionReadOnly.GetValueOrDefault(c.state) {
 		var logger *slog.Logger
 		var ro *spanner.ReadOnlyTransaction
 		var bo *spanner.BatchReadOnlyTransaction
-		if batchReadOnly {
+		if propertyTransactionBatchReadOnly.GetValueOrDefault(c.state) {
 			logger = c.logger.With("tx", "batchro")
 			var err error
 			// BatchReadOnly transactions (currently) do not support inline-begin.
 			// This means that the transaction options must be supplied here, and not through a callback.
-			bo, err = c.client.BatchReadOnlyTransaction(ctx, batchReadOnlyTxOpts.TimestampBound)
+			bo, err = c.client.BatchReadOnlyTransaction(c.tx.ctx, propertyReadOnlyStaleness.GetValueOrDefault(c.state))
 			if err != nil {
 				return nil, err
 			}
 			ro = &bo.ReadOnlyTransaction
 		} else {
 			logger = c.logger.With("tx", "ro")
-			ro = c.client.ReadOnlyTransaction().WithBeginTransactionOption(readOnlyTxOpts.BeginTransactionOption)
+			beginTxOpt := c.convertDefaultBeginTransactionOption(propertyBeginTransactionOption.GetValueOrDefault(c.state))
+			ro = c.client.ReadOnlyTransaction().WithBeginTransactionOption(beginTxOpt)
 		}
-		c.tx = &readOnlyTransaction{
+		return &readOnlyTransaction{
 			roTx:   ro,
 			boTx:   bo,
 			logger: logger,
-			close: func(result txResult) {
-				closeFunc()
-				if result == txResultCommit {
-					_ = c.state.Commit()
-				} else {
-					_ = c.state.Rollback()
-				}
-				c.tx = nil
-			},
+			close:  closeFunc,
 			timestampBoundCallback: func() spanner.TimestampBound {
 				return propertyReadOnlyStaleness.GetValueOrDefault(c.state)
 			},
-		}
-		return c.tx, nil
+		}, nil
 	}
 
-	// These options are only used to determine how to start the transaction.
-	// All other options are fetched in a callback that is called when the transaction is actually started.
-	// That callback reads all transaction options from the connection state at that moment. This allows
-	// applications to execute a series of statement like this:
-	// BEGIN TRANSACTION;
-	// SET LOCAL transaction_tag='my_tag';
-	// SET LOCAL commit_priority=LOW;
-	// INSERT INTO my_table ... -- This starts the transaction with the options above included.
 	opts := spanner.TransactionOptions{}
 	opts.BeginTransactionOption = c.convertDefaultBeginTransactionOption(propertyBeginTransactionOption.GetValueOrDefault(c.state))
 
-	tx, err := spanner.NewReadWriteStmtBasedTransactionWithCallbackForOptions(ctx, c.client, opts, func() spanner.TransactionOptions {
+	tx, err := spanner.NewReadWriteStmtBasedTransactionWithCallbackForOptions(c.tx.ctx, c.client, opts, func() spanner.TransactionOptions {
 		defer func() {
 			// Reset the transaction_tag after starting the transaction.
 			_ = propertyTransactionTag.ResetValue(c.state, connectionstate.ContextUser)
 		}()
-		return c.effectiveTransactionOptions(isolationLevelFromTxOpts, c.options( /*reset=*/ true))
+		return c.effectiveTransactionOptions(spannerpb.TransactionOptions_ISOLATION_LEVEL_UNSPECIFIED, c.options( /*reset=*/ true))
 	})
 	if err != nil {
 		return nil, err
 	}
 	logger := c.logger.With("tx", "rw")
-	c.tx = &readWriteTransaction{
-		ctx:    ctx,
+	return &readWriteTransaction{
+		ctx:    c.tx.ctx,
 		conn:   c,
 		logger: logger,
 		rwTx:   tx,
 		close: func(result txResult, commitResponse *spanner.CommitResponse, commitErr error) {
-			closeFunc()
 			c.prevTx = c.tx
-			c.tx = nil
 			if commitErr == nil {
 				c.setCommitResponse(commitResponse)
-				if result == txResultCommit {
-					_ = c.state.Commit()
-				} else {
-					_ = c.state.Rollback()
-				}
+				closeFunc(result)
 			} else {
-				_ = c.state.Rollback()
+				closeFunc(txResultRollback)
 			}
 		},
-		// Disable internal retries if any of these options have been set.
 		retryAborts: sync.OnceValue(func() bool {
-			return c.RetryAbortsInternally() && !disableRetryAborts
+			return c.RetryAbortsInternally()
 		}),
-	}
-	c.clearCommitResponse()
-	return c.tx, nil
+	}, nil
 }
 
 func (c *conn) effectiveTransactionOptions(isolationLevelFromTxOpts spannerpb.TransactionOptions_IsolationLevel, execOptions *ExecOptions) spanner.TransactionOptions {
@@ -1340,22 +1329,6 @@ func (c *conn) inTransaction() bool {
 	return c.tx != nil
 }
 
-func (c *conn) inReadOnlyTransaction() bool {
-	if c.tx != nil {
-		_, ok := c.tx.(*readOnlyTransaction)
-		return ok
-	}
-	return false
-}
-
-func (c *conn) inReadWriteTransaction() bool {
-	if c.tx != nil {
-		_, ok := c.tx.(*readWriteTransaction)
-		return ok
-	}
-	return false
-}
-
 // Commit is not part of the public API of the database/sql driver.
 // It is exported for internal reasons, and may receive breaking changes without prior notice.
 //