chore: remove transaction id from api

Author: olavloite

conn.go

@@ -694,6 +694,17 @@ func sum(affected []int64) int64 {
 	return sum
 }
 
+func (c *conn) WriteMutations(ctx context.Context, ms []*spanner.Mutation) (*spanner.CommitResponse, error) {
+	if c.inTransaction() {
+		return nil, c.BufferWrite(ms)
+	}
+	ts, err := c.Apply(ctx, ms)
+	if err != nil {
+		return nil, err
+	}
+	return &spanner.CommitResponse{CommitTs: ts}, nil
+}
+
 func (c *conn) Apply(ctx context.Context, ms []*spanner.Mutation, opts ...spanner.ApplyOption) (commitTimestamp time.Time, err error) {
 	if c.inTransaction() {
 		return time.Time{}, spanner.ToSpannerError(
@@ -1091,6 +1102,26 @@ func (c *conn) getBatchReadOnlyTransactionOptions() BatchReadOnlyTransactionOpti
 	return BatchReadOnlyTransactionOptions{TimestampBound: c.ReadOnlyStaleness()}
 }
 
+func (c *conn) BeginReadOnlyTransaction(ctx context.Context, options *ReadOnlyTransactionOptions) (driver.Tx, error) {
+	c.withTempReadOnlyTransactionOptions(options)
+	tx, err := c.BeginTx(ctx, driver.TxOptions{ReadOnly: true})
+	if err != nil {
+		c.withTempReadOnlyTransactionOptions(nil)
+		return nil, err
+	}
+	return tx, nil
+}
+
+func (c *conn) BeginReadWriteTransaction(ctx context.Context, options *ReadWriteTransactionOptions) (driver.Tx, error) {
+	c.withTempTransactionOptions(options)
+	tx, err := c.BeginTx(ctx, driver.TxOptions{})
+	if err != nil {
+		c.withTempTransactionOptions(nil)
+		return nil, err
+	}
+	return tx, nil
+}
+
 func (c *conn) Begin() (driver.Tx, error) {
 	return c.BeginTx(context.Background(), driver.TxOptions{})
 }
@@ -1274,18 +1305,21 @@ func (c *conn) inReadWriteTransaction() bool {
 	return false
 }
 
-func (c *conn) commit(ctx context.Context) (*spanner.CommitResponse, error) {
+func (c *conn) Commit(ctx context.Context) (*spanner.CommitResponse, error) {
 	if !c.inTransaction() {
 		return nil, status.Errorf(codes.FailedPrecondition, "this connection does not have a transaction")
 	}
 	// TODO: Pass in context to the tx.Commit() function.
 	if err := c.tx.Commit(); err != nil {
 		return nil, err
 	}
-	return c.CommitResponse()
+
+	// This will return either the commit response or nil, depending on whether the transaction was a
+	// read/write transaction or a read-only transaction.
+	return propertyCommitResponse.GetValueOrDefault(c.state), nil
 }
 
-func (c *conn) rollback(ctx context.Context) error {
+func (c *conn) Rollback(ctx context.Context) error {
 	if !c.inTransaction() {
 		return status.Errorf(codes.FailedPrecondition, "this connection does not have a transaction")
 	}

spannerlib/api/connection.go

@@ -3,11 +3,11 @@ package api
 import (
 	"context"
 	"database/sql"
+	"database/sql/driver"
 	"fmt"
 	"strings"
 	"sync"
 	"sync/atomic"
-	"time"
 
 	"cloud.google.com/go/spanner"
 	"cloud.google.com/go/spanner/apiv1/spannerpb"
@@ -26,18 +26,18 @@ func CloseConnection(poolId, connId int64) error {
 	return conn.close()
 }
 
-func Apply(poolId, connId int64, mutations *spannerpb.BatchWriteRequest_MutationGroup) (*spannerpb.CommitResponse, error) {
+func WriteMutations(poolId, connId int64, mutations *spannerpb.BatchWriteRequest_MutationGroup) (*spannerpb.CommitResponse, error) {
 	conn, err := findConnection(poolId, connId)
 	if err != nil {
 		return nil, err
 	}
-	return conn.apply(mutations)
+	return conn.writeMutations(mutations)
 }
 
-func BeginTransaction(poolId, connId int64, txOpts *spannerpb.TransactionOptions) (int64, error) {
+func BeginTransaction(poolId, connId int64, txOpts *spannerpb.TransactionOptions) error {
 	conn, err := findConnection(poolId, connId)
 	if err != nil {
-		return 0, err
+		return err
 	}
 	return conn.BeginTransaction(txOpts)
 }
@@ -62,12 +62,18 @@ type Connection struct {
 	results    *sync.Map
 	resultsIdx atomic.Int64
 
-	transactions    *sync.Map
-	transactionsIdx atomic.Int64
-
 	backend *sql.Conn
 }
 
+// spannerConn is an internal interface that contains the internal functions that are used by this API.
+type spannerConn interface {
+	WriteMutations(ctx context.Context, ms []*spanner.Mutation) (*spanner.CommitResponse, error)
+	BeginReadOnlyTransaction(ctx context.Context, options *spannerdriver.ReadOnlyTransactionOptions) (driver.Tx, error)
+	BeginReadWriteTransaction(ctx context.Context, options *spannerdriver.ReadWriteTransactionOptions) (driver.Tx, error)
+	Commit(ctx context.Context) (*spanner.CommitResponse, error)
+	Rollback(ctx context.Context) error
+}
+
 type queryExecutor interface {
 	ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
 	QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
@@ -79,19 +85,14 @@ func (conn *Connection) close() error {
 		_ = res.Close()
 		return true
 	})
-	conn.transactions.Range(func(key, value interface{}) bool {
-		res := value.(*transaction)
-		_ = res.Close()
-		return true
-	})
 	err := conn.backend.Close()
 	if err != nil {
 		return err
 	}
 	return nil
 }
 
-func (conn *Connection) apply(mutation *spannerpb.BatchWriteRequest_MutationGroup) (*spannerpb.CommitResponse, error) {
+func (conn *Connection) writeMutations(mutation *spannerpb.BatchWriteRequest_MutationGroup) (*spannerpb.CommitResponse, error) {
 	ctx := context.Background()
 	mutations := make([]*spanner.Mutation, 0, len(mutation.Mutations))
 	for _, m := range mutation.Mutations {
@@ -101,47 +102,59 @@ func (conn *Connection) apply(mutation *spannerpb.BatchWriteRequest_MutationGrou
 		}
 		mutations = append(mutations, spannerMutation)
 	}
-	var commitTimestamp time.Time
+	var commitResponse *spanner.CommitResponse
 	if err := conn.backend.Raw(func(driverConn any) (err error) {
-		spannerConn, _ := driverConn.(spannerdriver.SpannerConn)
-		commitTimestamp, err = spannerConn.Apply(ctx, mutations)
+		sc, _ := driverConn.(spannerConn)
+		commitResponse, err = sc.WriteMutations(ctx, mutations)
 		return err
 	}); err != nil {
 		return nil, err
 	}
+
+	// The commit response is nil if the connection is currently in a transaction.
+	if commitResponse == nil {
+		return nil, nil
+	}
 	response := spannerpb.CommitResponse{
-		CommitTimestamp: timestamppb.New(commitTimestamp),
+		CommitTimestamp: timestamppb.New(commitResponse.CommitTs),
 	}
 	return &response, nil
 }
 
-func (conn *Connection) BeginTransaction(txOpts *spannerpb.TransactionOptions) (int64, error) {
-	var tx *sql.Tx
+func (conn *Connection) BeginTransaction(txOpts *spannerpb.TransactionOptions) error {
 	var err error
+	ctx := context.Background()
 	if txOpts.GetReadOnly() != nil {
-		tx, err = spannerdriver.BeginReadOnlyTransactionOnConn(
-			context.Background(), conn.backend, convertToReadOnlyOpts(txOpts))
+		return conn.beginReadOnlyTransaction(ctx, convertToReadOnlyOpts(txOpts))
 	} else if txOpts.GetPartitionedDml() != nil {
 		err = spanner.ToSpannerError(status.Error(codes.InvalidArgument, "transaction type not supported"))
 	} else {
-		tx, err = spannerdriver.BeginReadWriteTransactionOnConn(
-			context.Background(), conn.backend, convertToReadWriteTransactionOptions(txOpts))
+		return conn.beginReadWriteTransaction(ctx, convertToReadWriteTransactionOptions(txOpts))
 	}
 	if err != nil {
-		return 0, err
-	}
-	id := conn.transactionsIdx.Add(1)
-	res := &transaction{
-		backend: tx,
-		conn:    conn,
-		txOpts:  txOpts,
+		return err
 	}
-	conn.transactions.Store(id, res)
-	return id, nil
+	return nil
+}
+
+func (conn *Connection) beginReadOnlyTransaction(ctx context.Context, opts *spannerdriver.ReadOnlyTransactionOptions) error {
+	return conn.backend.Raw(func(driverConn any) (err error) {
+		sc, _ := driverConn.(spannerConn)
+		_, err = sc.BeginReadOnlyTransaction(ctx, opts)
+		return err
+	})
+}
+
+func (conn *Connection) beginReadWriteTransaction(ctx context.Context, opts *spannerdriver.ReadWriteTransactionOptions) error {
+	return conn.backend.Raw(func(driverConn any) (err error) {
+		sc, _ := driverConn.(spannerConn)
+		_, err = sc.BeginReadWriteTransaction(ctx, opts)
+		return err
+	})
 }
 
-func convertToReadOnlyOpts(txOpts *spannerpb.TransactionOptions) spannerdriver.ReadOnlyTransactionOptions {
-	return spannerdriver.ReadOnlyTransactionOptions{
+func convertToReadOnlyOpts(txOpts *spannerpb.TransactionOptions) *spannerdriver.ReadOnlyTransactionOptions {
+	return &spannerdriver.ReadOnlyTransactionOptions{
 		TimestampBound: convertTimestampBound(txOpts),
 	}
 }
@@ -162,12 +175,12 @@ func convertTimestampBound(txOpts *spannerpb.TransactionOptions) spanner.Timesta
 	return spanner.TimestampBound{}
 }
 
-func convertToReadWriteTransactionOptions(txOpts *spannerpb.TransactionOptions) spannerdriver.ReadWriteTransactionOptions {
+func convertToReadWriteTransactionOptions(txOpts *spannerpb.TransactionOptions) *spannerdriver.ReadWriteTransactionOptions {
 	readLockMode := spannerpb.TransactionOptions_ReadWrite_READ_LOCK_MODE_UNSPECIFIED
 	if txOpts.GetReadWrite() != nil {
 		readLockMode = txOpts.GetReadWrite().GetReadLockMode()
 	}
-	return spannerdriver.ReadWriteTransactionOptions{
+	return &spannerdriver.ReadWriteTransactionOptions{
 		TransactionOptions: spanner.TransactionOptions{
 			IsolationLevel: txOpts.GetIsolationLevel(),
 			ReadLockMode:   readLockMode,

spannerlib/api/pool.go

@@ -69,9 +69,8 @@ func CreateConnection(poolId int64) (int64, error) {
 	}
 	id := poolsIdx.Add(1)
 	conn := &Connection{
-		backend:      sqlConn,
-		results:      &sync.Map{},
-		transactions: &sync.Map{},
+		backend: sqlConn,
+		results: &sync.Map{},
 	}
 	pool.connections.Store(id, conn)
 
@@ -104,16 +103,3 @@ func findRows(poolId, connId, rowsId int64) (*rows, error) {
 	res := r.(*rows)
 	return res, nil
 }
-
-func findTx(poolId, connId, txId int64) (*transaction, error) {
-	conn, err := findConnection(poolId, connId)
-	if err != nil {
-		return nil, err
-	}
-	r, ok := conn.transactions.Load(txId)
-	if !ok {
-		return nil, fmt.Errorf("tx %v not found", txId)
-	}
-	res := r.(*transaction)
-	return res, nil
-}

spannerlib/api/transaction.go

@@ -1,6 +1,7 @@
 package api
 
 import (
+	"context"
 	"database/sql"
 
 	"cloud.google.com/go/spanner"
@@ -9,46 +10,20 @@ import (
 	"google.golang.org/protobuf/types/known/timestamppb"
 )
 
-func BufferWrite(poolId, connId, txId int64, mutations *spannerpb.BatchWriteRequest_MutationGroup) error {
-	tx, err := findTx(poolId, connId, txId)
-	if err != nil {
-		return err
-	}
-	return tx.bufferWrite(mutations)
-}
-
-func ExecuteTransaction(poolId, connId, txId int64, request *spannerpb.ExecuteSqlRequest) (int64, error) {
-	tx, err := findTx(poolId, connId, txId)
-	if err != nil {
-		return 0, err
-	}
-	return tx.Execute(request)
-}
-
-func Commit(poolId, connId, txId int64) (*spannerpb.CommitResponse, error) {
-	tx, err := findTx(poolId, connId, txId)
-	if err != nil {
-		return nil, err
-	}
+func Commit(poolId, connId int64) (*spannerpb.CommitResponse, error) {
 	conn, err := findConnection(poolId, connId)
 	if err != nil {
 		return nil, err
 	}
-	conn.transactions.Delete(txId)
-	return tx.Commit()
+	return commit(conn)
 }
 
-func Rollback(poolId, connId, txId int64) error {
-	tx, err := findTx(poolId, connId, txId)
-	if err != nil {
-		return err
-	}
+func Rollback(poolId, connId int64) error {
 	conn, err := findConnection(poolId, connId)
 	if err != nil {
 		return err
 	}
-	conn.transactions.Delete(txId)
-	return tx.Rollback()
+	return rollback(conn)
 }
 
 type transaction struct {
@@ -91,33 +66,30 @@ func (tx *transaction) Execute(statement *spannerpb.ExecuteSqlRequest) (int64, e
 	return execute(tx.conn, tx.backend, statement)
 }
 
-func (tx *transaction) Commit() (*spannerpb.CommitResponse, error) {
-	tx.closed = true
-	if err := tx.backend.Commit(); err != nil {
-		return nil, err
-	}
+func commit(conn *Connection) (*spannerpb.CommitResponse, error) {
 	var response *spanner.CommitResponse
-	if tx.txOpts.GetReadWrite() == nil {
-		return &spannerpb.CommitResponse{}, nil
-	}
-	if err := tx.conn.backend.Raw(func(driverConn any) (err error) {
-		spannerConn, _ := driverConn.(spannerdriver.SpannerConn)
-		response, err = spannerConn.CommitResponse()
+	if err := conn.backend.Raw(func(driverConn any) (err error) {
+		spannerConn, _ := driverConn.(spannerConn)
+		response, err = spannerConn.Commit(context.Background())
 		if err != nil {
 			return err
 		}
 		return nil
 	}); err != nil {
 		return nil, err
 	}
+
+	// The commit response is nil for read-only transactions.
+	if response == nil {
+		return nil, nil
+	}
 	// TODO: Include commit stats
 	return &spannerpb.CommitResponse{CommitTimestamp: timestamppb.New(response.CommitTs)}, nil
 }
 
-func (tx *transaction) Rollback() error {
-	tx.closed = true
-	if err := tx.backend.Rollback(); err != nil {
-		return err
-	}
-	return nil
+func rollback(conn *Connection) error {
+	return conn.backend.Raw(func(driverConn any) (err error) {
+		spannerConn, _ := driverConn.(spannerConn)
+		return spannerConn.Rollback(context.Background())
+	})
 }