You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@seata.apache.org by te...@apache.org on 2026/08/25 08:19:35 UTC
(incubator-seata-go) branch master updated: fix(sql): consolidate rowsWithStmt close handling (#1151)
This is an automated email from the ASF dual-hosted git repository.
thunguo pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-seata-go.git
The following commit(s) were added to refs/heads/master by this push:
new 46fd409d fix(sql): consolidate rowsWithStmt close handling (#1151)
46fd409d is described below
commit 46fd409d6d6daf488b14d426b7162725de5f7492
Author: Mochimia <27...@qq.com>
AuthorDate: Tue Aug 25 16:19:29 2026 +0800
fix(sql): consolidate rowsWithStmt close handling (#1151)
* fix(sql): consolidate rowsWithStmt close handling
Signed-off-by: Mochimia <27...@qq.com>
* test(sql): cover rowsWithStmt prepare fallbacks
Signed-off-by: Mochimia <27...@qq.com>
---------
Signed-off-by: Mochimia <27...@qq.com>
Co-authored-by: ThunGuo <te...@apache.org>
---
pkg/datasource/sql/conn_at.go | 17 +--
pkg/datasource/sql/conn_at_test.go | 33 ++++++
pkg/datasource/sql/conn_xa.go | 3 +-
pkg/datasource/sql/exec/at/delete_executor.go | 2 +-
pkg/datasource/sql/exec/at/insert_executor.go | 17 +--
.../sql/exec/at/select_for_update_executor.go | 2 +-
.../sql/exec/at/select_for_update_executor_test.go | 115 +++++++++++++++++++++
pkg/datasource/sql/exec/at/update_executor.go | 4 +-
pkg/datasource/sql/util/ctxutil.go | 7 +-
pkg/datasource/sql/util/ctxutil_test.go | 43 ++++++++
10 files changed, 205 insertions(+), 38 deletions(-)
diff --git a/pkg/datasource/sql/conn_at.go b/pkg/datasource/sql/conn_at.go
index 6fe27860..b5fd4f63 100644
--- a/pkg/datasource/sql/conn_at.go
+++ b/pkg/datasource/sql/conn_at.go
@@ -32,12 +32,6 @@ import (
"seata.apache.org/seata-go/v2/pkg/util/log"
)
-// rowsWithStmt wraps driver.Rows and closes the statement when rows are closed
-type rowsWithStmt struct {
- driver.Rows
- stmt driver.Stmt
-}
-
// nonRetryableATError preserves the commit error without exposing the retry signal to database/sql.
type nonRetryableATError struct{ cause error }
@@ -49,15 +43,6 @@ func (e nonRetryableATError) Is(target error) bool {
func (e nonRetryableATError) As(target any) bool { return errors.As(e.cause, target) }
-func (r *rowsWithStmt) Close() error {
- rowsErr := r.Rows.Close()
- stmtErr := r.stmt.Close()
- if rowsErr != nil {
- return rowsErr
- }
- return stmtErr
-}
-
// ATConn Database connection proxy object under XA transaction model
// Conn is assumed to be stateful.
type ATConn struct {
@@ -194,7 +179,7 @@ func (c *ATConn) QueryContext(ctx context.Context, query string, args []driver.N
}
// Wrap rows with statement to close both together
- wrappedRows := &rowsWithStmt{Rows: rows, stmt: stmt}
+ wrappedRows := util.NewRowsWithStmt(rows, stmt)
return types.NewResult(types.WithRows(wrappedRows)), nil
}
diff --git a/pkg/datasource/sql/conn_at_test.go b/pkg/datasource/sql/conn_at_test.go
index 69a1c41d..2d56be84 100644
--- a/pkg/datasource/sql/conn_at_test.go
+++ b/pkg/datasource/sql/conn_at_test.go
@@ -147,6 +147,39 @@ func TestATConnAllowsPreparedMultiSQLForPostgreSQL(t *testing.T) {
assert.NoError(t, err)
}
+func TestATConnQueryContextPrepareFallbackPreservesCloseErrors(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ targetConn := mock.NewMockTestDriverConn(ctrl)
+ targetStmt := mock.NewMockTestDriverStmt(ctrl)
+ targetRows := mock.NewMockTestDriverRows(ctrl)
+ ctx := context.Background()
+ query := "SELECT id FROM t_user WHERE id = ?"
+ args := []driver.NamedValue{{Ordinal: 1, Value: int64(1)}}
+ rowsCloseErr := errors.New("rows close failed")
+ stmtCloseErr := errors.New("statement close failed")
+
+ targetConn.EXPECT().QueryContext(ctx, query, args).Return(nil, driver.ErrSkip)
+ targetConn.EXPECT().Prepare(query).Return(targetStmt, nil)
+ targetStmt.EXPECT().QueryContext(ctx, args).Return(targetRows, nil)
+ targetRows.EXPECT().Close().Return(rowsCloseErr)
+ targetStmt.EXPECT().Close().Return(stmtCloseErr)
+
+ conn := &ATConn{Conn: &Conn{
+ res: &DBResource{dbType: types.DBTypeMySQL},
+ txCtx: types.NewTxCtx(),
+ targetConn: targetConn,
+ dbType: types.DBTypeMySQL,
+ }}
+
+ rows, err := conn.QueryContext(ctx, query, args)
+ if !assert.NoError(t, err) || !assert.NotNil(t, rows) {
+ return
+ }
+ closeErr := rows.Close()
+ assert.ErrorIs(t, closeErr, rowsCloseErr)
+ assert.ErrorIs(t, closeErr, stmtCloseErr)
+}
+
type postgresMockRows struct {
columns []string
data [][]driver.Value
diff --git a/pkg/datasource/sql/conn_xa.go b/pkg/datasource/sql/conn_xa.go
index a7c73cf6..71c98235 100644
--- a/pkg/datasource/sql/conn_xa.go
+++ b/pkg/datasource/sql/conn_xa.go
@@ -26,6 +26,7 @@ import (
"time"
"seata.apache.org/seata-go/v2/pkg/datasource/sql/types"
+ "seata.apache.org/seata-go/v2/pkg/datasource/sql/util"
"seata.apache.org/seata-go/v2/pkg/datasource/sql/xa"
"seata.apache.org/seata-go/v2/pkg/tm"
"seata.apache.org/seata-go/v2/pkg/util/log"
@@ -375,7 +376,7 @@ func (c *XAConn) queryPreparedInBranch(ctx context.Context, query string, args [
_ = stmt.Close()
return nil, err
}
- return types.NewResult(types.WithRows(&rowsWithStmt{Rows: rows, stmt: stmt})), nil
+ return types.NewResult(types.WithRows(util.NewRowsWithStmt(rows, stmt))), nil
}
// xaDeferredCommitTx wraps an XA branch tx whose commit is deferred until the
diff --git a/pkg/datasource/sql/exec/at/delete_executor.go b/pkg/datasource/sql/exec/at/delete_executor.go
index 9d848ed6..dde78bba 100644
--- a/pkg/datasource/sql/exec/at/delete_executor.go
+++ b/pkg/datasource/sql/exec/at/delete_executor.go
@@ -122,7 +122,7 @@ func (d *deleteExecutor) beforeImage(ctx context.Context) (*types.RecordImage, e
}
// Wrap rows with statement to close both together
- rowsi = &rowsWithStmt{Rows: rowsi, stmt: stmt}
+ rowsi = util.NewRowsWithStmt(rowsi, stmt)
}
defer func() {
diff --git a/pkg/datasource/sql/exec/at/insert_executor.go b/pkg/datasource/sql/exec/at/insert_executor.go
index 0bb77068..c1b9e791 100644
--- a/pkg/datasource/sql/exec/at/insert_executor.go
+++ b/pkg/datasource/sql/exec/at/insert_executor.go
@@ -245,21 +245,6 @@ func (i *insertExecutor) buildPostgreSQLReturningInsertSQL(meta *types.TableMeta
return trimTrailingSemicolon(i.execContext.Query) + " RETURNING " + strings.Join(returningColumns, ", "), nil
}
-// rowsWithStmt wraps driver.Rows and closes the statement when rows are closed
-type rowsWithStmt struct {
- driver.Rows
- stmt driver.Stmt
-}
-
-func (r *rowsWithStmt) Close() error {
- rowsErr := r.Rows.Close()
- stmtErr := r.stmt.Close()
- if rowsErr != nil {
- return rowsErr
- }
- return stmtErr
-}
-
func (i *insertExecutor) queryRows(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {
// Try direct query first
queryerCtx, ok := i.execContext.Conn.(driver.QueryerContext)
@@ -307,7 +292,7 @@ func (i *insertExecutor) queryRows(ctx context.Context, query string, args []dri
return nil, err
}
- return &rowsWithStmt{Rows: rows, stmt: stmt}, nil
+ return util.NewRowsWithStmt(rows, stmt), nil
}
func namedValuesToValues(named []driver.NamedValue) ([]driver.Value, error) {
diff --git a/pkg/datasource/sql/exec/at/select_for_update_executor.go b/pkg/datasource/sql/exec/at/select_for_update_executor.go
index 91ddf34d..d2e9f87f 100644
--- a/pkg/datasource/sql/exec/at/select_for_update_executor.go
+++ b/pkg/datasource/sql/exec/at/select_for_update_executor.go
@@ -401,5 +401,5 @@ func (s *selectForUpdateExecutor) exec(ctx context.Context, sql string, nvdargs
return nil, nil
}
- return &rowsWithStmt{Rows: rows, stmt: stmt}, nil
+ return util.NewRowsWithStmt(rows, stmt), nil
}
diff --git a/pkg/datasource/sql/exec/at/select_for_update_executor_test.go b/pkg/datasource/sql/exec/at/select_for_update_executor_test.go
index 36afe65c..731ead1a 100644
--- a/pkg/datasource/sql/exec/at/select_for_update_executor_test.go
+++ b/pkg/datasource/sql/exec/at/select_for_update_executor_test.go
@@ -18,12 +18,14 @@
package at
import (
+ "context"
"database/sql/driver"
"io"
"testing"
"github.com/stretchr/testify/assert"
+ "seata.apache.org/seata-go/v2/pkg/datasource/sql/datasource"
"seata.apache.org/seata-go/v2/pkg/datasource/sql/parser"
"seata.apache.org/seata-go/v2/pkg/datasource/sql/types"
)
@@ -167,6 +169,119 @@ func TestBuildLockKey(t *testing.T) {
assert.Equal(t, "t_user:1_oid11,2_oid22,3_oid33", lockKey)
}
+func TestPrepareFallbackRowsCloseStatement(t *testing.T) {
+ ctx := context.Background()
+ updateSQL := "UPDATE t_user SET name = ? WHERE id = ?"
+ deleteSQL := "DELETE FROM t_user WHERE id = ?"
+ updateParser, err := parser.DoParser(updateSQL)
+ assert.NoError(t, err)
+ deleteParser, err := parser.DoParser(deleteSQL)
+ assert.NoError(t, err)
+
+ meta := &types.TableMeta{
+ TableName: "t_user",
+ ColumnNames: []string{"id", "name"},
+ Columns: map[string]types.ColumnMeta{
+ "id": {ColumnName: "id", DatabaseTypeString: "BIGINT"},
+ "name": {ColumnName: "name", DatabaseTypeString: "VARCHAR"},
+ },
+ Indexs: map[string]types.IndexMeta{
+ "PRIMARY": {
+ IType: types.IndexTypePrimaryKey,
+ ColumnName: "id",
+ Columns: []types.ColumnMeta{{ColumnName: "id"}},
+ },
+ },
+ }
+ datasource.RegisterTableCache(types.DBTypeMySQL, &stubTableMetaCache{meta: meta})
+ updateArgs := []driver.NamedValue{{Ordinal: 1, Value: "updated"}, {Ordinal: 2, Value: int64(1)}}
+ deleteArgs := []driver.NamedValue{{Ordinal: 1, Value: int64(1)}}
+ beforeImage := types.RecordImage{Rows: []types.RowImage{{Columns: []types.ColumnImage{{ColumnName: "id", Value: int64(1)}}}}}
+
+ tests := []struct {
+ name string
+ run func(*prepareFallbackConn) error
+ }{
+ {name: "insert query", run: func(conn *prepareFallbackConn) error {
+ rows, err := (&insertExecutor{execContext: &types.ExecContext{Conn: conn}}).queryRows(ctx, "SELECT 1", nil)
+ if err != nil {
+ return err
+ }
+ return rows.Close()
+ }},
+ {name: "select for update", run: func(conn *prepareFallbackConn) error {
+ rows, err := (&selectForUpdateExecutor{execContext: &types.ExecContext{Conn: conn}}).exec(ctx, "SELECT 1 FOR UPDATE", nil, nil)
+ if err != nil {
+ return err
+ }
+ return rows.Close()
+ }},
+ {name: "update before image", run: func(conn *prepareFallbackConn) error {
+ executor := &updateExecutor{parserCtx: updateParser, execContext: &types.ExecContext{
+ Conn: conn, Query: updateSQL, NamedValues: updateArgs, TxCtx: types.NewTxCtx(),
+ }}
+ _, err := executor.beforeImage(ctx)
+ return err
+ }},
+ {name: "update after image", run: func(conn *prepareFallbackConn) error {
+ executor := &updateExecutor{parserCtx: updateParser, execContext: &types.ExecContext{Conn: conn}}
+ _, err := executor.afterImage(ctx, beforeImage)
+ return err
+ }},
+ {name: "delete before image", run: func(conn *prepareFallbackConn) error {
+ executor := &deleteExecutor{parserCtx: deleteParser, execContext: &types.ExecContext{
+ Conn: conn, Query: deleteSQL, NamedValues: deleteArgs, TxCtx: types.NewTxCtx(),
+ }}
+ _, err := executor.beforeImage(ctx)
+ return err
+ }},
+ }
+
+ for _, tt := range tests {
+ t.Run(tt.name, func(t *testing.T) {
+ conn := &prepareFallbackConn{}
+ assert.NoError(t, tt.run(conn))
+ assert.True(t, conn.rowsClosed)
+ assert.True(t, conn.stmtClosed)
+ })
+ }
+}
+
+type prepareFallbackConn struct {
+ rowsClosed bool
+ stmtClosed bool
+}
+
+func (c *prepareFallbackConn) Prepare(string) (driver.Stmt, error) {
+ return &prepareFallbackStmt{conn: c}, nil
+}
+
+func (*prepareFallbackConn) Close() error { return nil }
+func (*prepareFallbackConn) Begin() (driver.Tx, error) { return nil, nil }
+func (*prepareFallbackConn) QueryContext(context.Context, string, []driver.NamedValue) (driver.Rows, error) {
+ return nil, driver.ErrSkip
+}
+
+type prepareFallbackStmt struct{ conn *prepareFallbackConn }
+
+func (s *prepareFallbackStmt) Close() error { s.conn.stmtClosed = true; return nil }
+func (*prepareFallbackStmt) NumInput() int { return -1 }
+func (*prepareFallbackStmt) Exec([]driver.Value) (driver.Result, error) {
+ return nil, driver.ErrSkip
+}
+func (s *prepareFallbackStmt) Query([]driver.Value) (driver.Rows, error) {
+ return &prepareFallbackRows{conn: s.conn}, nil
+}
+func (s *prepareFallbackStmt) QueryContext(context.Context, []driver.NamedValue) (driver.Rows, error) {
+ return &prepareFallbackRows{conn: s.conn}, nil
+}
+
+type prepareFallbackRows struct{ conn *prepareFallbackConn }
+
+func (*prepareFallbackRows) Columns() []string { return nil }
+func (r *prepareFallbackRows) Close() error { r.conn.rowsClosed = true; return nil }
+func (*prepareFallbackRows) Next([]driver.Value) error { return io.EOF }
+
type mockRows struct{}
func (m mockRows) Columns() []string {
diff --git a/pkg/datasource/sql/exec/at/update_executor.go b/pkg/datasource/sql/exec/at/update_executor.go
index 67bbc3c8..a5f179a9 100644
--- a/pkg/datasource/sql/exec/at/update_executor.go
+++ b/pkg/datasource/sql/exec/at/update_executor.go
@@ -148,7 +148,7 @@ func (u *updateExecutor) beforeImage(ctx context.Context) (*types.RecordImage, e
}
// Wrap rows with statement to close both together
- rowsi = &rowsWithStmt{Rows: rowsi, stmt: stmt}
+ rowsi = util.NewRowsWithStmt(rowsi, stmt)
}
defer func() {
@@ -228,7 +228,7 @@ func (u *updateExecutor) afterImage(ctx context.Context, beforeImage types.Recor
}
// Wrap rows with statement to close both together
- rowsi = &rowsWithStmt{Rows: rowsi, stmt: stmt}
+ rowsi = util.NewRowsWithStmt(rowsi, stmt)
}
defer func() {
diff --git a/pkg/datasource/sql/util/ctxutil.go b/pkg/datasource/sql/util/ctxutil.go
index e5beae68..23538540 100644
--- a/pkg/datasource/sql/util/ctxutil.go
+++ b/pkg/datasource/sql/util/ctxutil.go
@@ -127,6 +127,11 @@ type rowsWithStmt struct {
stmt driver.Stmt
}
+// NewRowsWithStmt wraps rows and closes both rows and stmt when the returned rows are closed.
+func NewRowsWithStmt(rows driver.Rows, stmt driver.Stmt) driver.Rows {
+ return &rowsWithStmt{Rows: rows, stmt: stmt}
+}
+
func (r *rowsWithStmt) Close() error {
var rowsErr error
if r.Rows != nil {
@@ -208,5 +213,5 @@ func CtxDriverQueryWithPrepareFallback(ctx context.Context, conn driver.Conn, qu
return nil, err
}
- return &rowsWithStmt{Rows: rows, stmt: stmt}, nil
+ return NewRowsWithStmt(rows, stmt), nil
}
diff --git a/pkg/datasource/sql/util/ctxutil_test.go b/pkg/datasource/sql/util/ctxutil_test.go
index 415bd266..e4f2a33a 100644
--- a/pkg/datasource/sql/util/ctxutil_test.go
+++ b/pkg/datasource/sql/util/ctxutil_test.go
@@ -533,6 +533,49 @@ func TestCtxDriverExecWithPrepareFallback(t *testing.T) {
assert.Equal(t, int64(1), affected)
}
+func TestNewRowsWithStmt(t *testing.T) {
+ t.Run("close preserves rows and statement errors", func(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+
+ targetRows := mock.NewMockTestDriverRows(ctrl)
+ targetStmt := mock.NewMockTestDriverStmt(ctrl)
+ rowsCloseErr := errors.New("rows close failed")
+ stmtCloseErr := errors.New("statement close failed")
+
+ targetRows.EXPECT().Close().Times(1).Return(rowsCloseErr)
+ targetStmt.EXPECT().Close().Times(1).Return(stmtCloseErr)
+
+ closeErr := NewRowsWithStmt(targetRows, targetStmt).Close()
+ assert.ErrorIs(t, closeErr, rowsCloseErr)
+ assert.ErrorIs(t, closeErr, stmtCloseErr)
+ })
+
+ t.Run("nil rows", func(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+
+ targetStmt := mock.NewMockTestDriverStmt(ctrl)
+ targetStmt.EXPECT().Close().Times(1).Return(nil)
+
+ assert.NoError(t, NewRowsWithStmt(nil, targetStmt).Close())
+ })
+
+ t.Run("nil statement", func(t *testing.T) {
+ ctrl := gomock.NewController(t)
+ defer ctrl.Finish()
+
+ targetRows := mock.NewMockTestDriverRows(ctrl)
+ targetRows.EXPECT().Close().Times(1).Return(nil)
+
+ assert.NoError(t, NewRowsWithStmt(targetRows, nil).Close())
+ })
+
+ t.Run("nil rows and statement", func(t *testing.T) {
+ assert.NoError(t, NewRowsWithStmt(nil, nil).Close())
+ })
+}
+
func TestCtxDriverQueryWithPrepareFallback(t *testing.T) {
ctx := context.Background()
query := "SELECT id, name FROM t_user WHERE id IN (?, ?)"
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org