From 406b6bf410147f7331f44f6754079a4ef1184286 Mon Sep 17 00:00:00 2001 From: jflyfox Date: Fri, 21 May 2021 15:30:21 +0800 Subject: [PATCH] infract internal link --- database/gdb/gdb.go | 42 ++-- database/gdb/gdb_core.go | 181 +++-------------- database/gdb/gdb_core_link.go | 31 +++ ...transaction.go => gdb_core_transaction.go} | 57 +++--- database/gdb/gdb_core_underlying.go | 186 ++++++++++++++++++ database/gdb/gdb_core_utility.go | 24 ++- database/gdb/gdb_driver_mssql.go | 10 +- database/gdb/gdb_driver_mysql.go | 10 +- database/gdb/gdb_driver_oracle.go | 12 +- database/gdb/gdb_driver_pgsql.go | 10 +- database/gdb/gdb_driver_sqlite.go | 10 +- database/gdb/gdb_model_utility.go | 10 +- database/gdb/gdb_statement.go | 18 +- database/gdb/gdb_z_mysql_transaction_test.go | 120 ++++++++++- 14 files changed, 465 insertions(+), 256 deletions(-) create mode 100644 database/gdb/gdb_core_link.go rename database/gdb/{gdb_transaction.go => gdb_core_transaction.go} (95%) create mode 100644 database/gdb/gdb_core_underlying.go diff --git a/database/gdb/gdb.go b/database/gdb/gdb.go index 8d634359e..5b18d5639 100644 --- a/database/gdb/gdb.go +++ b/database/gdb/gdb.go @@ -113,8 +113,8 @@ type DB interface { // Master/Slave specification support. // =========================================================================== - Master() (*sql.DB, error) // See Core.Master. - Slave() (*sql.DB, error) // See Core.Slave. + Master(schema ...string) (*sql.DB, error) // See Core.Master. + Slave(schema ...string) (*sql.DB, error) // See Core.Slave. // =========================================================================== // Ping-Pong. @@ -187,16 +187,28 @@ type Driver interface { New(core *Core, node *ConfigNode) (DB, error) } +// Link is a common database function wrapper interface. +type Link interface { + Query(sql string, args ...interface{}) (*sql.Rows, error) + Exec(sql string, args ...interface{}) (sql.Result, error) + Prepare(sql string) (*sql.Stmt, error) + QueryContext(ctx context.Context, sql string, args ...interface{}) (*sql.Rows, error) + ExecContext(ctx context.Context, sql string, args ...interface{}) (sql.Result, error) + PrepareContext(ctx context.Context, sql string) (*sql.Stmt, error) + IsTransaction() bool +} + // Sql is the sql recording struct. type Sql struct { - Sql string // SQL string(may contain reserved char '?'). - Type string // SQL operation type. - Args []interface{} // Arguments for this sql. - Format string // Formatted sql which contains arguments in the sql. - Error error // Execution result. - Start int64 // Start execution timestamp in milliseconds. - End int64 // End execution timestamp in milliseconds. - Group string // Group is the group name of the configuration that the sql is executed from. + Sql string // SQL string(may contain reserved char '?'). + Type string // SQL operation type. + Args []interface{} // Arguments for this sql. + Format string // Formatted sql which contains arguments in the sql. + Error error // Execution result. + Start int64 // Start execution timestamp in milliseconds. + End int64 // End execution timestamp in milliseconds. + Group string // Group is the group name of the configuration that the sql is executed from. + IsTransaction bool // IsTransaction marks whether this sql is executed in transaction. } // TableField is the struct for table field. @@ -211,16 +223,6 @@ type TableField struct { Comment string // Comment. } -// Link is a common database function wrapper interface. -type Link interface { - Query(sql string, args ...interface{}) (*sql.Rows, error) - Exec(sql string, args ...interface{}) (sql.Result, error) - Prepare(sql string) (*sql.Stmt, error) - QueryContext(ctx context.Context, sql string, args ...interface{}) (*sql.Rows, error) - ExecContext(ctx context.Context, sql string, args ...interface{}) (sql.Result, error) - PrepareContext(ctx context.Context, sql string) (*sql.Stmt, error) -} - // Counter is the type for update count. type Counter struct { Field string diff --git a/database/gdb/gdb_core.go b/database/gdb/gdb_core.go index 14a14917c..9c0786935 100644 --- a/database/gdb/gdb_core.go +++ b/database/gdb/gdb_core.go @@ -19,7 +19,6 @@ import ( "github.com/gogf/gf/text/gstr" "github.com/gogf/gf/container/gvar" - "github.com/gogf/gf/os/gtime" "github.com/gogf/gf/text/gregex" "github.com/gogf/gf/util/gconv" ) @@ -94,162 +93,26 @@ func (c *Core) GetCtxTimeout(timeoutType int, ctx context.Context) (context.Cont // Master creates and returns a connection from master node if master-slave configured. // It returns the default connection if master-slave not configured. -func (c *Core) Master() (*sql.DB, error) { - return c.getSqlDb(true, c.schema.Val()) +func (c *Core) Master(schema ...string) (*sql.DB, error) { + useSchema := "" + if len(schema) > 0 && schema[0] != "" { + useSchema = schema[0] + } else { + useSchema = c.schema.Val() + } + return c.getSqlDb(true, useSchema) } // Slave creates and returns a connection from slave node if master-slave configured. // It returns the default connection if master-slave not configured. -func (c *Core) Slave() (*sql.DB, error) { - return c.getSqlDb(false, c.schema.Val()) -} - -// Query commits one query SQL to underlying driver and returns the execution result. -// It is most commonly used for data querying. -func (c *Core) Query(sql string, args ...interface{}) (rows *sql.Rows, err error) { - link, err := c.db.Slave() - if err != nil { - return nil, err - } - return c.DoQuery(c.GetCtx(), link, sql, args...) -} - -// DoQuery commits the sql string and its arguments to underlying driver -// through given link object and returns the execution result. -func (c *Core) DoQuery(ctx context.Context, link Link, sql string, args ...interface{}) (rows *sql.Rows, err error) { - sql, args = formatSql(sql, args) - sql, args = c.db.HandleSqlBeforeCommit(ctx, link, sql, args) - if c.GetConfig().QueryTimeout > 0 { - ctx, _ = context.WithTimeout(ctx, c.GetConfig().QueryTimeout) - } - mTime1 := gtime.TimestampMilli() - rows, err = link.QueryContext(ctx, sql, args...) - mTime2 := gtime.TimestampMilli() - sqlObj := &Sql{ - Sql: sql, - Type: "DB.QueryContext", - Args: args, - Format: FormatSqlWithArgs(sql, args), - Error: err, - Start: mTime1, - End: mTime2, - Group: c.db.GetGroup(), - } - // Tracing and logging. - c.addSqlToTracing(ctx, sqlObj) - if c.db.GetDebug() { - c.writeSqlToLogger(ctx, sqlObj) - } - if err == nil { - return rows, nil +func (c *Core) Slave(schema ...string) (*sql.DB, error) { + useSchema := "" + if len(schema) > 0 && schema[0] != "" { + useSchema = schema[0] } else { - err = formatError(err, sql, args...) + useSchema = c.schema.Val() } - return nil, err -} - -// Exec commits one query SQL to underlying driver and returns the execution result. -// It is most commonly used for data inserting and updating. -func (c *Core) Exec(sql string, args ...interface{}) (result sql.Result, err error) { - link, err := c.db.Master() - if err != nil { - return nil, err - } - return c.DoExec(c.GetCtx(), link, sql, args...) -} - -// DoExec commits the sql string and its arguments to underlying driver -// through given link object and returns the execution result. -func (c *Core) DoExec(ctx context.Context, link Link, sql string, args ...interface{}) (result sql.Result, err error) { - sql, args = formatSql(sql, args) - sql, args = c.db.HandleSqlBeforeCommit(ctx, link, sql, args) - if c.GetConfig().ExecTimeout > 0 { - var cancelFunc context.CancelFunc - ctx, cancelFunc = context.WithTimeout(ctx, c.GetConfig().ExecTimeout) - defer cancelFunc() - } - - mTime1 := gtime.TimestampMilli() - if !c.db.GetDryRun() { - result, err = link.ExecContext(ctx, sql, args...) - } else { - result = new(SqlResult) - } - mTime2 := gtime.TimestampMilli() - sqlObj := &Sql{ - Sql: sql, - Type: "DB.ExecContext", - Args: args, - Format: FormatSqlWithArgs(sql, args), - Error: err, - Start: mTime1, - End: mTime2, - Group: c.db.GetGroup(), - } - // Tracing and logging. - c.addSqlToTracing(ctx, sqlObj) - if c.db.GetDebug() { - c.writeSqlToLogger(ctx, sqlObj) - } - return result, formatError(err, sql, args...) -} - -// Prepare creates a prepared statement for later queries or executions. -// Multiple queries or executions may be run concurrently from the -// returned statement. -// The caller must call the statement's Close method -// when the statement is no longer needed. -// -// The parameter `execOnMaster` specifies whether executing the sql on master node, -// or else it executes the sql on slave node if master-slave configured. -func (c *Core) Prepare(sql string, execOnMaster ...bool) (*Stmt, error) { - var ( - err error - link Link - ) - if len(execOnMaster) > 0 && execOnMaster[0] { - if link, err = c.db.Master(); err != nil { - return nil, err - } - } else { - if link, err = c.db.Slave(); err != nil { - return nil, err - } - } - return c.DoPrepare(c.GetCtx(), link, sql) -} - -// DoPrepare calls prepare function on given link object and returns the statement object. -func (c *Core) DoPrepare(ctx context.Context, link Link, sql string) (*Stmt, error) { - if c.GetConfig().PrepareTimeout > 0 { - // DO NOT USE cancel function in prepare statement. - ctx, _ = context.WithTimeout(ctx, c.GetConfig().PrepareTimeout) - } - var ( - mTime1 = gtime.TimestampMilli() - stmt, err = link.PrepareContext(ctx, sql) - mTime2 = gtime.TimestampMilli() - sqlObj = &Sql{ - Sql: sql, - Type: "DB.PrepareContext", - Args: nil, - Format: FormatSqlWithArgs(sql, nil), - Error: err, - Start: mTime1, - End: mTime2, - Group: c.db.GetGroup(), - } - ) - // Tracing and logging. - c.addSqlToTracing(ctx, sqlObj) - if c.db.GetDebug() { - c.writeSqlToLogger(ctx, sqlObj) - } - return &Stmt{ - Stmt: stmt, - core: c, - sql: sql, - }, err + return c.getSqlDb(false, useSchema) } // GetAll queries and returns data records from database. @@ -260,7 +123,7 @@ func (c *Core) GetAll(sql string, args ...interface{}) (Result, error) { // DoGetAll queries and returns data records from database. func (c *Core) DoGetAll(ctx context.Context, link Link, sql string, args ...interface{}) (result Result, err error) { if link == nil { - link, err = c.db.Slave() + link, err = c.SlaveLink() if err != nil { return nil, err } @@ -536,7 +399,7 @@ func (c *Core) DoInsert(ctx context.Context, link Link, table string, data inter updateStr = fmt.Sprintf("ON DUPLICATE KEY UPDATE %s", updateStr) } if link == nil { - if link, err = c.db.Master(); err != nil { + if link, err = c.MasterLink(); err != nil { return nil, err } } @@ -641,7 +504,7 @@ func (c *Core) DoBatchInsert(ctx context.Context, link Link, table string, list return result, gerror.New("data list cannot be empty") } if link == nil { - if link, err = c.db.Master(); err != nil { + if link, err = c.MasterLink(); err != nil { return } } @@ -800,7 +663,7 @@ func (c *Core) DoUpdate(ctx context.Context, link Link, table string, data inter } // If no link passed, it then uses the master link. if link == nil { - if link, err = c.db.Master(); err != nil { + if link, err = c.MasterLink(); err != nil { return nil, err } } @@ -826,7 +689,7 @@ func (c *Core) Delete(table string, condition interface{}, args ...interface{}) // This function is usually used for custom interface definition, you do not need call it manually. func (c *Core) DoDelete(ctx context.Context, link Link, table string, condition string, args ...interface{}) (result sql.Result, err error) { if link == nil { - if link, err = c.db.Master(); err != nil { + if link, err = c.MasterLink(); err != nil { return nil, err } } @@ -891,8 +754,10 @@ func (c *Core) MarshalJSON() ([]byte, error) { // It is enabled only if configuration "debug" is true. func (c *Core) writeSqlToLogger(ctx context.Context, sql *Sql) { var transactionIdStr string - if v := ctx.Value(transactionIdForLoggerCtx); v != nil { - transactionIdStr = fmt.Sprintf(`[%d] `, v.(uint64)) + if sql.IsTransaction { + if v := ctx.Value(transactionIdForLoggerCtx); v != nil { + transactionIdStr = fmt.Sprintf(`[%d] `, v.(uint64)) + } } s := fmt.Sprintf("[%3d ms] [%s] %s%s", sql.End-sql.Start, sql.Group, transactionIdStr, sql.Format) if sql.Error != nil { diff --git a/database/gdb/gdb_core_link.go b/database/gdb/gdb_core_link.go new file mode 100644 index 000000000..06ea3190b --- /dev/null +++ b/database/gdb/gdb_core_link.go @@ -0,0 +1,31 @@ +// Copyright GoFrame Author(https://goframe.org). All Rights Reserved. +// +// This Source Code Form is subject to the terms of the MIT License. +// If a copy of the MIT was not distributed with this file, +// You can obtain one at https://github.com/gogf/gf. + +package gdb + +import ( + "database/sql" +) + +// dbLink is used to implement interface Link for DB. +type dbLink struct { + *sql.DB +} + +// txLink is used to implement interface Link for TX. +type txLink struct { + *sql.Tx +} + +// IsTransaction returns if current Link is a transaction. +func (*dbLink) IsTransaction() bool { + return false +} + +// IsTransaction returns if current Link is a transaction. +func (*txLink) IsTransaction() bool { + return true +} diff --git a/database/gdb/gdb_transaction.go b/database/gdb/gdb_core_transaction.go similarity index 95% rename from database/gdb/gdb_transaction.go rename to database/gdb/gdb_core_transaction.go index abc46a7da..4636d9166 100644 --- a/database/gdb/gdb_transaction.go +++ b/database/gdb/gdb_core_transaction.go @@ -59,14 +59,15 @@ func (c *Core) doBeginCtx(ctx context.Context) (*TX, error) { rawTx, err = master.Begin() mTime2 = gtime.TimestampMilli() sqlObj = &Sql{ - Sql: sqlStr, - Type: "DB.Begin", - Args: nil, - Format: sqlStr, - Error: err, - Start: mTime1, - End: mTime2, - Group: c.db.GetGroup(), + Sql: sqlStr, + Type: "DB.Begin", + Args: nil, + Format: sqlStr, + Error: err, + Start: mTime1, + End: mTime2, + Group: c.db.GetGroup(), + IsTransaction: true, } ) if err == nil { @@ -198,14 +199,15 @@ func (tx *TX) Commit() error { err = tx.tx.Commit() mTime2 = gtime.TimestampMilli() sqlObj = &Sql{ - Sql: sqlStr, - Type: "TX.Commit", - Args: nil, - Format: sqlStr, - Error: err, - Start: mTime1, - End: mTime2, - Group: tx.db.GetGroup(), + Sql: sqlStr, + Type: "TX.Commit", + Args: nil, + Format: sqlStr, + Error: err, + Start: mTime1, + End: mTime2, + Group: tx.db.GetGroup(), + IsTransaction: true, } ) tx.db.GetCore().addSqlToTracing(tx.ctx, sqlObj) @@ -230,14 +232,15 @@ func (tx *TX) Rollback() error { err = tx.tx.Rollback() mTime2 = gtime.TimestampMilli() sqlObj = &Sql{ - Sql: sqlStr, - Type: "TX.Rollback", - Args: nil, - Format: sqlStr, - Error: err, - Start: mTime1, - End: mTime2, - Group: tx.db.GetGroup(), + Sql: sqlStr, + Type: "TX.Rollback", + Args: nil, + Format: sqlStr, + Error: err, + Start: mTime1, + End: mTime2, + Group: tx.db.GetGroup(), + IsTransaction: true, } ) tx.db.GetCore().addSqlToTracing(tx.ctx, sqlObj) @@ -314,13 +317,13 @@ func (tx *TX) Transaction(ctx context.Context, f func(ctx context.Context, tx *T // Query does query operation on transaction. // See Core.Query. func (tx *TX) Query(sql string, args ...interface{}) (rows *sql.Rows, err error) { - return tx.db.GetCore().DoQuery(tx.ctx, tx.tx, sql, args...) + return tx.db.GetCore().DoQuery(tx.ctx, &txLink{tx.tx}, sql, args...) } // Exec does none query operation on transaction. // See Core.Exec. func (tx *TX) Exec(sql string, args ...interface{}) (sql.Result, error) { - return tx.db.GetCore().DoExec(tx.ctx, tx.tx, sql, args...) + return tx.db.GetCore().DoExec(tx.ctx, &txLink{tx.tx}, sql, args...) } // Prepare creates a prepared statement for later queries or executions. @@ -329,7 +332,7 @@ func (tx *TX) Exec(sql string, args ...interface{}) (sql.Result, error) { // The caller must call the statement's Close method // when the statement is no longer needed. func (tx *TX) Prepare(sql string) (*Stmt, error) { - return tx.db.GetCore().DoPrepare(tx.ctx, tx.tx, sql) + return tx.db.GetCore().DoPrepare(tx.ctx, &txLink{tx.tx}, sql) } // GetAll queries and returns data records from database. diff --git a/database/gdb/gdb_core_underlying.go b/database/gdb/gdb_core_underlying.go new file mode 100644 index 000000000..316127df0 --- /dev/null +++ b/database/gdb/gdb_core_underlying.go @@ -0,0 +1,186 @@ +// Copyright GoFrame Author(https://goframe.org). All Rights Reserved. +// +// This Source Code Form is subject to the terms of the MIT License. +// If a copy of the MIT was not distributed with this file, +// You can obtain one at https://github.com/gogf/gf. +// + +package gdb + +import ( + "context" + "database/sql" + + "github.com/gogf/gf/os/gtime" +) + +// Query commits one query SQL to underlying driver and returns the execution result. +// It is most commonly used for data querying. +func (c *Core) Query(sql string, args ...interface{}) (rows *sql.Rows, err error) { + return c.DoQuery(c.GetCtx(), nil, sql, args...) +} + +// DoQuery commits the sql string and its arguments to underlying driver +// through given link object and returns the execution result. +func (c *Core) DoQuery(ctx context.Context, link Link, sql string, args ...interface{}) (rows *sql.Rows, err error) { + // Transaction checks. + if link == nil { + if link, err = c.SlaveLink(); err != nil { + return nil, err + } + } else if !link.IsTransaction() { + if tx := TXFromCtx(ctx, c.db.GetGroup()); tx != nil { + link = &txLink{tx.tx} + } + } + // Link execution. + sql, args = formatSql(sql, args) + sql, args = c.db.HandleSqlBeforeCommit(ctx, link, sql, args) + if c.GetConfig().QueryTimeout > 0 { + ctx, _ = context.WithTimeout(ctx, c.GetConfig().QueryTimeout) + } + mTime1 := gtime.TimestampMilli() + rows, err = link.QueryContext(ctx, sql, args...) + mTime2 := gtime.TimestampMilli() + sqlObj := &Sql{ + Sql: sql, + Type: "DB.QueryContext", + Args: args, + Format: FormatSqlWithArgs(sql, args), + Error: err, + Start: mTime1, + End: mTime2, + Group: c.db.GetGroup(), + IsTransaction: link.IsTransaction(), + } + // Tracing and logging. + c.addSqlToTracing(ctx, sqlObj) + if c.db.GetDebug() { + c.writeSqlToLogger(ctx, sqlObj) + } + if err == nil { + return rows, nil + } else { + err = formatError(err, sql, args...) + } + return nil, err +} + +// Exec commits one query SQL to underlying driver and returns the execution result. +// It is most commonly used for data inserting and updating. +func (c *Core) Exec(sql string, args ...interface{}) (result sql.Result, err error) { + return c.DoExec(c.GetCtx(), nil, sql, args...) +} + +// DoExec commits the sql string and its arguments to underlying driver +// through given link object and returns the execution result. +func (c *Core) DoExec(ctx context.Context, link Link, sql string, args ...interface{}) (result sql.Result, err error) { + // Transaction checks. + if link == nil { + if link, err = c.MasterLink(); err != nil { + return nil, err + } + } else if !link.IsTransaction() { + if tx := TXFromCtx(ctx, c.db.GetGroup()); tx != nil { + link = &txLink{tx.tx} + } + } + // Link execution. + sql, args = formatSql(sql, args) + sql, args = c.db.HandleSqlBeforeCommit(ctx, link, sql, args) + if c.GetConfig().ExecTimeout > 0 { + var cancelFunc context.CancelFunc + ctx, cancelFunc = context.WithTimeout(ctx, c.GetConfig().ExecTimeout) + defer cancelFunc() + } + + mTime1 := gtime.TimestampMilli() + if !c.db.GetDryRun() { + result, err = link.ExecContext(ctx, sql, args...) + } else { + result = new(SqlResult) + } + mTime2 := gtime.TimestampMilli() + sqlObj := &Sql{ + Sql: sql, + Type: "DB.ExecContext", + Args: args, + Format: FormatSqlWithArgs(sql, args), + Error: err, + Start: mTime1, + End: mTime2, + Group: c.db.GetGroup(), + IsTransaction: link.IsTransaction(), + } + // Tracing and logging. + c.addSqlToTracing(ctx, sqlObj) + if c.db.GetDebug() { + c.writeSqlToLogger(ctx, sqlObj) + } + return result, formatError(err, sql, args...) +} + +// Prepare creates a prepared statement for later queries or executions. +// Multiple queries or executions may be run concurrently from the +// returned statement. +// The caller must call the statement's Close method +// when the statement is no longer needed. +// +// The parameter `execOnMaster` specifies whether executing the sql on master node, +// or else it executes the sql on slave node if master-slave configured. +func (c *Core) Prepare(sql string, execOnMaster ...bool) (*Stmt, error) { + var ( + err error + link Link + ) + if len(execOnMaster) > 0 && execOnMaster[0] { + if link, err = c.MasterLink(); err != nil { + return nil, err + } + } else { + if link, err = c.SlaveLink(); err != nil { + return nil, err + } + } + return c.DoPrepare(c.GetCtx(), link, sql) +} + +// DoPrepare calls prepare function on given link object and returns the statement object. +func (c *Core) DoPrepare(ctx context.Context, link Link, sql string) (*Stmt, error) { + if link != nil && !link.IsTransaction() { + if tx := TXFromCtx(ctx, c.db.GetGroup()); tx != nil { + link = &txLink{tx.tx} + } + } + if c.GetConfig().PrepareTimeout > 0 { + // DO NOT USE cancel function in prepare statement. + ctx, _ = context.WithTimeout(ctx, c.GetConfig().PrepareTimeout) + } + var ( + mTime1 = gtime.TimestampMilli() + stmt, err = link.PrepareContext(ctx, sql) + mTime2 = gtime.TimestampMilli() + sqlObj = &Sql{ + Sql: sql, + Type: "DB.PrepareContext", + Args: nil, + Format: FormatSqlWithArgs(sql, nil), + Error: err, + Start: mTime1, + End: mTime2, + Group: c.db.GetGroup(), + IsTransaction: link.IsTransaction(), + } + ) + // Tracing and logging. + c.addSqlToTracing(ctx, sqlObj) + if c.db.GetDebug() { + c.writeSqlToLogger(ctx, sqlObj) + } + return &Stmt{ + Stmt: stmt, + core: c, + link: link, + sql: sql, + }, err +} diff --git a/database/gdb/gdb_core_utility.go b/database/gdb/gdb_core_utility.go index 3528d873e..0eaeb16c4 100644 --- a/database/gdb/gdb_core_utility.go +++ b/database/gdb/gdb_core_utility.go @@ -7,22 +7,26 @@ package gdb -import ( - "database/sql" -) - -// GetMaster acts like function Master but with additional `schema` parameter specifying +// MasterLink acts like function Master but with additional `schema` parameter specifying // the schema for the connection. It is defined for internal usage. // Also see Master. -func (c *Core) GetMaster(schema ...string) (*sql.DB, error) { - return c.getSqlDb(true, schema...) +func (c *Core) MasterLink(schema ...string) (Link, error) { + db, err := c.db.Master(schema...) + if err != nil { + return nil, err + } + return &dbLink{db}, nil } -// GetSlave acts like function Slave but with additional `schema` parameter specifying +// SlaveLink acts like function Slave but with additional `schema` parameter specifying // the schema for the connection. It is defined for internal usage. // Also see Slave. -func (c *Core) GetSlave(schema ...string) (*sql.DB, error) { - return c.getSqlDb(false, schema...) +func (c *Core) SlaveLink(schema ...string) (Link, error) { + db, err := c.db.Slave(schema...) + if err != nil { + return nil, err + } + return &dbLink{db}, nil } // QuoteWord checks given string `s` a word, if true quotes it with security chars of the database diff --git a/database/gdb/gdb_driver_mssql.go b/database/gdb/gdb_driver_mssql.go index fde84e762..d0bb93cfb 100644 --- a/database/gdb/gdb_driver_mssql.go +++ b/database/gdb/gdb_driver_mssql.go @@ -187,7 +187,7 @@ func (d *DriverMssql) parseSql(sql string) string { // It's mainly used in cli tool chain for automatically generating the models. func (d *DriverMssql) Tables(ctx context.Context, schema ...string) (tables []string, err error) { var result Result - link, err := d.GetSlave(schema...) + link, err := d.SlaveLink(schema...) if err != nil { return nil, err } @@ -213,20 +213,20 @@ func (d *DriverMssql) TableFields(ctx context.Context, link Link, table string, if gstr.Contains(table, " ") { return nil, gerror.New("function TableFields supports only single table operations") } - checkSchema := d.db.GetSchema() + useSchema := d.db.GetSchema() if len(schema) > 0 && schema[0] != "" { - checkSchema = schema[0] + useSchema = schema[0] } tableFieldsCacheKey := fmt.Sprintf( `mssql_table_fields_%s_%s@group:%s`, - table, checkSchema, d.GetGroup(), + table, useSchema, d.GetGroup(), ) v := tableFieldsMap.GetOrSetFuncLock(tableFieldsCacheKey, func() interface{} { var ( result Result ) if link == nil { - link, err = d.GetSlave(checkSchema) + link, err = d.SlaveLink(useSchema) if err != nil { return nil } diff --git a/database/gdb/gdb_driver_mysql.go b/database/gdb/gdb_driver_mysql.go index ee87ecd94..724b353ec 100644 --- a/database/gdb/gdb_driver_mysql.go +++ b/database/gdb/gdb_driver_mysql.go @@ -85,7 +85,7 @@ func (d *DriverMysql) HandleSqlBeforeCommit(ctx context.Context, link Link, sql // It's mainly used in cli tool chain for automatically generating the models. func (d *DriverMysql) Tables(ctx context.Context, schema ...string) (tables []string, err error) { var result Result - link, err := d.GetSlave(schema...) + link, err := d.SlaveLink(schema...) if err != nil { return nil, err } @@ -119,20 +119,20 @@ func (d *DriverMysql) TableFields(ctx context.Context, link Link, table string, if gstr.Contains(table, " ") { return nil, gerror.New("function TableFields supports only single table operations") } - checkSchema := d.schema.Val() + useSchema := d.schema.Val() if len(schema) > 0 && schema[0] != "" { - checkSchema = schema[0] + useSchema = schema[0] } tableFieldsCacheKey := fmt.Sprintf( `mysql_table_fields_%s_%s@group:%s`, - table, checkSchema, d.GetGroup(), + table, useSchema, d.GetGroup(), ) v := tableFieldsMap.GetOrSetFuncLock(tableFieldsCacheKey, func() interface{} { var ( result Result ) if link == nil { - link, err = d.GetSlave(checkSchema) + link, err = d.SlaveLink(useSchema) if err != nil { return nil } diff --git a/database/gdb/gdb_driver_oracle.go b/database/gdb/gdb_driver_oracle.go index b067de9b4..be356f176 100644 --- a/database/gdb/gdb_driver_oracle.go +++ b/database/gdb/gdb_driver_oracle.go @@ -189,13 +189,13 @@ func (d *DriverOracle) TableFields(ctx context.Context, link Link, table string, if gstr.Contains(table, " ") { return nil, gerror.New("function TableFields supports only single table operations") } - checkSchema := d.db.GetSchema() + useSchema := d.db.GetSchema() if len(schema) > 0 && schema[0] != "" { - checkSchema = schema[0] + useSchema = schema[0] } tableFieldsCacheKey := fmt.Sprintf( `oracle_table_fields_%s_%s@group:%s`, - table, checkSchema, d.GetGroup(), + table, useSchema, d.GetGroup(), ) v := tableFieldsMap.GetOrSetFuncLock(tableFieldsCacheKey, func() interface{} { var ( @@ -213,7 +213,7 @@ FROM USER_TAB_COLUMNS WHERE TABLE_NAME = '%s' ORDER BY COLUMN_ID`, ) structureSql, _ = gregex.ReplaceString(`[\n\r\s]+`, " ", gstr.Trim(structureSql)) if link == nil { - link, err = d.GetSlave(checkSchema) + link, err = d.SlaveLink(useSchema) if err != nil { return nil } @@ -340,7 +340,7 @@ func (d *DriverOracle) DoInsert(ctx context.Context, link Link, table string, da } if link == nil { - if link, err = d.db.Master(); err != nil { + if link, err = d.MasterLink(); err != nil { return nil, err } } @@ -416,7 +416,7 @@ func (d *DriverOracle) DoBatchInsert(ctx context.Context, link Link, table strin return result, gerror.New("empty data list") } if link == nil { - if link, err = d.db.Master(); err != nil { + if link, err = d.MasterLink(); err != nil { return } } diff --git a/database/gdb/gdb_driver_pgsql.go b/database/gdb/gdb_driver_pgsql.go index 2d0bd76f6..8fb6ad243 100644 --- a/database/gdb/gdb_driver_pgsql.go +++ b/database/gdb/gdb_driver_pgsql.go @@ -92,7 +92,7 @@ func (d *DriverPgsql) HandleSqlBeforeCommit(ctx context.Context, link Link, sql // It's mainly used in cli tool chain for automatically generating the models. func (d *DriverPgsql) Tables(ctx context.Context, schema ...string) (tables []string, err error) { var result Result - link, err := d.GetSlave(schema...) + link, err := d.SlaveLink(schema...) if err != nil { return nil, err } @@ -122,13 +122,13 @@ func (d *DriverPgsql) TableFields(ctx context.Context, link Link, table string, return nil, gerror.New("function TableFields supports only single table operations") } table, _ = gregex.ReplaceString("\"", "", table) - checkSchema := d.db.GetSchema() + useSchema := d.db.GetSchema() if len(schema) > 0 && schema[0] != "" { - checkSchema = schema[0] + useSchema = schema[0] } tableFieldsCacheKey := fmt.Sprintf( `pgsql_table_fields_%s_%s@group:%s`, - table, checkSchema, d.GetGroup(), + table, useSchema, d.GetGroup(), ) v := tableFieldsMap.GetOrSetFuncLock(tableFieldsCacheKey, func() interface{} { var ( @@ -143,7 +143,7 @@ ORDER BY a.attnum`, ) structureSql, _ = gregex.ReplaceString(`[\n\r\s]+`, " ", gstr.Trim(structureSql)) if link == nil { - link, err = d.GetSlave(checkSchema) + link, err = d.SlaveLink(useSchema) if err != nil { return nil } diff --git a/database/gdb/gdb_driver_sqlite.go b/database/gdb/gdb_driver_sqlite.go index c78a3eb27..69f4a42a5 100644 --- a/database/gdb/gdb_driver_sqlite.go +++ b/database/gdb/gdb_driver_sqlite.go @@ -77,7 +77,7 @@ func (d *DriverSqlite) HandleSqlBeforeCommit(ctx context.Context, link Link, sql // It's mainly used in cli tool chain for automatically generating the models. func (d *DriverSqlite) Tables(ctx context.Context, schema ...string) (tables []string, err error) { var result Result - link, err := d.GetSlave(schema...) + link, err := d.SlaveLink(schema...) if err != nil { return nil, err } @@ -103,20 +103,20 @@ func (d *DriverSqlite) TableFields(ctx context.Context, link Link, table string, if gstr.Contains(table, " ") { return nil, gerror.New("function TableFields supports only single table operations") } - checkSchema := d.db.GetSchema() + useSchema := d.db.GetSchema() if len(schema) > 0 && schema[0] != "" { - checkSchema = schema[0] + useSchema = schema[0] } tableFieldsCacheKey := fmt.Sprintf( `sqlite_table_fields_%s_%s@group:%s`, - table, checkSchema, d.GetGroup(), + table, useSchema, d.GetGroup(), ) v := tableFieldsMap.GetOrSetFuncLock(tableFieldsCacheKey, func() interface{} { var ( result Result ) if link == nil { - link, err = d.GetSlave(checkSchema) + link, err = d.SlaveLink(useSchema) if err != nil { return nil } diff --git a/database/gdb/gdb_model_utility.go b/database/gdb/gdb_model_utility.go index 4dcd80a66..9efee81cf 100644 --- a/database/gdb/gdb_model_utility.go +++ b/database/gdb/gdb_model_utility.go @@ -28,9 +28,9 @@ func (m *Model) TableFields(table string, schema ...string) (fields map[string]* link Link ) if m.tx != nil { - link = m.tx.tx + link = &txLink{m.tx.tx} } else { - link, err = m.db.GetCore().GetSlave(schema...) + link, err = m.db.GetCore().SlaveLink(schema...) if err != nil { return } @@ -176,7 +176,7 @@ func (m *Model) doMappingAndFilterForInsertOrUpdateDataMap(data Map, allowOmitEm // The parameter `master` specifies whether using the master node if master-slave configured. func (m *Model) getLink(master bool) Link { if m.tx != nil { - return m.tx.tx + return &txLink{m.tx.tx} } linkType := m.linkType if linkType == 0 { @@ -188,13 +188,13 @@ func (m *Model) getLink(master bool) Link { } switch linkType { case linkTypeMaster: - link, err := m.db.GetCore().GetMaster(m.schema) + link, err := m.db.GetCore().MasterLink(m.schema) if err != nil { panic(err) } return link case linkTypeSlave: - link, err := m.db.GetCore().GetSlave(m.schema) + link, err := m.db.GetCore().SlaveLink(m.schema) if err != nil { panic(err) } diff --git a/database/gdb/gdb_statement.go b/database/gdb/gdb_statement.go index cc74e858c..81f14affe 100644 --- a/database/gdb/gdb_statement.go +++ b/database/gdb/gdb_statement.go @@ -26,6 +26,7 @@ import ( type Stmt struct { *sql.Stmt core *Core + link Link sql string } @@ -63,14 +64,15 @@ func (s *Stmt) doStmtCommit(stmtType string, ctx context.Context, args ...interf var ( timestampMilli2 = gtime.TimestampMilli() sqlObj = &Sql{ - Sql: s.sql, - Type: stmtType, - Args: args, - Format: FormatSqlWithArgs(s.sql, args), - Error: err, - Start: timestampMilli1, - End: timestampMilli2, - Group: s.core.db.GetGroup(), + Sql: s.sql, + Type: stmtType, + Args: args, + Format: FormatSqlWithArgs(s.sql, args), + Error: err, + Start: timestampMilli1, + End: timestampMilli2, + Group: s.core.db.GetGroup(), + IsTransaction: s.link.IsTransaction(), } ) // Tracing and logging. diff --git a/database/gdb/gdb_z_mysql_transaction_test.go b/database/gdb/gdb_z_mysql_transaction_test.go index b22b2a765..906bc6aca 100644 --- a/database/gdb/gdb_z_mysql_transaction_test.go +++ b/database/gdb/gdb_z_mysql_transaction_test.go @@ -833,7 +833,10 @@ func Test_Transaction_Nested_Begin_Rollback_Commit(t *testing.T) { func Test_Transaction_Nested_TX_Transaction_UseTX(t *testing.T) { table := createTable() defer dropTable(table) + db.SetDebug(true) + defer db.SetDebug(false) + gtest.C(t, func(t *gtest.T) { var ( err error @@ -887,17 +890,75 @@ func Test_Transaction_Nested_TX_Transaction_UseTX(t *testing.T) { }) t.AssertNil(err) - all, err := db.Model(table).All() + all, err := db.Ctx(ctx).Model(table).All() t.AssertNil(err) t.Assert(len(all), 1) t.Assert(all[0]["id"], 1) + + // another record. + err = db.Transaction(ctx, func(ctx context.Context, tx *gdb.TX) error { + // commit + err = tx.Transaction(ctx, func(ctx context.Context, tx *gdb.TX) error { + err = tx.Transaction(ctx, func(ctx context.Context, tx *gdb.TX) error { + err = tx.Transaction(ctx, func(ctx context.Context, tx *gdb.TX) error { + err = tx.Transaction(ctx, func(ctx context.Context, tx *gdb.TX) error { + err = tx.Transaction(ctx, func(ctx context.Context, tx *gdb.TX) error { + _, err = tx.Model(table).Data(g.Map{ + "id": 3, + "passport": "USER_1", + "password": "PASS_1", + "nickname": "NAME_1", + "create_time": gtime.Now().String(), + }).Insert() + t.AssertNil(err) + return err + }) + t.AssertNil(err) + return err + }) + t.AssertNil(err) + return err + }) + t.AssertNil(err) + return err + }) + t.AssertNil(err) + return err + }) + t.AssertNil(err) + // rollback + err = tx.Transaction(ctx, func(ctx context.Context, tx *gdb.TX) error { + _, err = tx.Model(table).Data(g.Map{ + "id": 4, + "passport": "USER_2", + "password": "PASS_2", + "nickname": "NAME_2", + "create_time": gtime.Now().String(), + }).Insert() + t.AssertNil(err) + panic("error") + return err + }) + t.AssertNE(err, nil) + return nil + }) + t.AssertNil(err) + + all, err = db.Ctx(ctx).Model(table).All() + t.AssertNil(err) + t.Assert(len(all), 2) + t.Assert(all[0]["id"], 1) + t.Assert(all[1]["id"], 3) }) } func Test_Transaction_Nested_TX_Transaction_UseDB(t *testing.T) { table := createTable() defer dropTable(table) + db.SetDebug(true) + defer db.SetDebug(false) + gtest.C(t, func(t *gtest.T) { var ( err error @@ -952,11 +1013,66 @@ func Test_Transaction_Nested_TX_Transaction_UseDB(t *testing.T) { return nil }) t.AssertNil(err) - all, err := db.Model(table).All() t.AssertNil(err) t.Assert(len(all), 1) t.Assert(all[0]["id"], 1) + + err = db.Transaction(ctx, func(ctx context.Context, tx *gdb.TX) error { + // commit + err = db.Transaction(ctx, func(ctx context.Context, tx *gdb.TX) error { + err = db.Transaction(ctx, func(ctx context.Context, tx *gdb.TX) error { + err = db.Transaction(ctx, func(ctx context.Context, tx *gdb.TX) error { + err = db.Transaction(ctx, func(ctx context.Context, tx *gdb.TX) error { + err = db.Transaction(ctx, func(ctx context.Context, tx *gdb.TX) error { + _, err = db.Model(table).Ctx(ctx).Data(g.Map{ + "id": 3, + "passport": "USER_1", + "password": "PASS_1", + "nickname": "NAME_1", + "create_time": gtime.Now().String(), + }).Insert() + t.AssertNil(err) + return err + }) + t.AssertNil(err) + return err + }) + t.AssertNil(err) + return err + }) + t.AssertNil(err) + return err + }) + t.AssertNil(err) + return err + }) + t.AssertNil(err) + + // rollback + err = db.Transaction(ctx, func(ctx context.Context, tx *gdb.TX) error { + _, err = tx.Model(table).Ctx(ctx).Data(g.Map{ + "id": 4, + "passport": "USER_2", + "password": "PASS_2", + "nickname": "NAME_2", + "create_time": gtime.Now().String(), + }).Insert() + t.AssertNil(err) + // panic makes this transaction rollback. + panic("error") + return err + }) + t.AssertNE(err, nil) + return nil + }) + t.AssertNil(err) + + all, err = db.Model(table).All() + t.AssertNil(err) + t.Assert(len(all), 2) + t.Assert(all[0]["id"], 1) + t.Assert(all[1]["id"], 3) }) }