// Copyright 2011 The Go Authors. All rights reserved. // Use of this source code is governed by a BSD-style // license that can be found in the LICENSE file. // Package sql provides a generic interface around SQL (or SQL-like) // databases. // // The sql package must be used in conjunction with a database driver. // See https://golang.org/s/sqldrivers for a list of drivers. // // Drivers that do not support context cancelation will not return until // after the query is completed. // // For usage examples, see the wiki page at // https://golang.org/s/sqlwiki. package sql import ( "context" "database/sql/driver" "errors" "fmt" "io" "reflect" "runtime" "sort" "sync" "sync/atomic" "time" ) var ( driversMu sync.RWMutex drivers = make(map[string]driver.Driver) ) // nowFunc returns the current time; it's overridden in tests. var nowFunc = time.Now // Register makes a database driver available by the provided name. // If Register is called twice with the same name or if driver is nil, // it panics. func Register(name string, driver driver.Driver) { driversMu.Lock() defer driversMu.Unlock() if driver == nil { panic("sql: Register driver is nil") } if _, dup := drivers[name]; dup { panic("sql: Register called twice for driver " + name) } drivers[name] = driver } func unregisterAllDrivers() { driversMu.Lock() defer driversMu.Unlock() // For tests. drivers = make(map[string]driver.Driver) } // Drivers returns a sorted list of the names of the registered drivers. func Drivers() []string { driversMu.RLock() defer driversMu.RUnlock() var list []string for name := range drivers { list = append(list, name) } sort.Strings(list) return list } // A NamedArg is a named argument. NamedArg values may be used as // arguments to Query or Exec and bind to the corresponding named // parameter in the SQL statement. // // For a more concise way to create NamedArg values, see // the Named function. type NamedArg struct { _Named_Fields_Required struct{} // Name is the name of the parameter placeholder. // // If empty, the ordinal position in the argument list will be // used. // // Name must omit any symbol prefix. Name string // Value is the value of the parameter. // It may be assigned the same value types as the query // arguments. Value interface{} } // Named provides a more concise way to create NamedArg values. // // Example usage: // // db.ExecContext(ctx, ` // delete from Invoice // where // TimeCreated < @end // and TimeCreated >= @start;`, // sql.Named("start", startTime), // sql.Named("end", endTime), // ) func Named(name string, value interface{}) NamedArg { // This method exists because the go1compat promise // doesn't guarantee that structs don't grow more fields, // so unkeyed struct literals are a vet error. Thus, we don't // want to allow sql.NamedArg{name, value}. return NamedArg{Name: name, Value: value} } // IsolationLevel is the transaction isolation level used in TxOptions. type IsolationLevel int // Various isolation levels that drivers may support in BeginTx. // If a driver does not support a given isolation level an error may be returned. // // See https://en.wikipedia.org/wiki/Isolation_(database_systems)#Isolation_levels. const ( LevelDefault IsolationLevel = iota LevelReadUncommitted LevelReadCommitted LevelWriteCommitted LevelRepeatableRead LevelSnapshot LevelSerializable LevelLinearizable ) // TxOptions holds the transaction options to be used in DB.BeginTx. type TxOptions struct { // Isolation is the transaction isolation level. // If zero, the driver or database's default level is used. Isolation IsolationLevel ReadOnly bool } // RawBytes is a byte slice that holds a reference to memory owned by // the database itself. After a Scan into a RawBytes, the slice is only // valid until the next call to Next, Scan, or Close. type RawBytes []byte // NullString represents a string that may be null. // NullString implements the Scanner interface so // it can be used as a scan destination: // // var s NullString // err := db.QueryRow("SELECT name FROM foo WHERE id=?", id).Scan(&s) // ... // if s.Valid { // // use s.String // } else { // // NULL value // } // type NullString struct { String string Valid bool // Valid is true if String is not NULL } // Scan implements the Scanner interface. func (ns *NullString) Scan(value interface{}) error { if value == nil { ns.String, ns.Valid = "", false return nil } ns.Valid = true return convertAssign(&ns.String, value) } // Value implements the driver Valuer interface. func (ns NullString) Value() (driver.Value, error) { if !ns.Valid { return nil, nil } return ns.String, nil } // NullInt64 represents an int64 that may be null. // NullInt64 implements the Scanner interface so // it can be used as a scan destination, similar to NullString. type NullInt64 struct { Int64 int64 Valid bool // Valid is true if Int64 is not NULL } // Scan implements the Scanner interface. func (n *NullInt64) Scan(value interface{}) error { if value == nil { n.Int64, n.Valid = 0, false return nil } n.Valid = true return convertAssign(&n.Int64, value) } // Value implements the driver Valuer interface. func (n NullInt64) Value() (driver.Value, error) { if !n.Valid { return nil, nil } return n.Int64, nil } // NullFloat64 represents a float64 that may be null. // NullFloat64 implements the Scanner interface so // it can be used as a scan destination, similar to NullString. type NullFloat64 struct { Float64 float64 Valid bool // Valid is true if Float64 is not NULL } // Scan implements the Scanner interface. func (n *NullFloat64) Scan(value interface{}) error { if value == nil { n.Float64, n.Valid = 0, false return nil } n.Valid = true return convertAssign(&n.Float64, value) } // Value implements the driver Valuer interface. func (n NullFloat64) Value() (driver.Value, error) { if !n.Valid { return nil, nil } return n.Float64, nil } // NullBool represents a bool that may be null. // NullBool implements the Scanner interface so // it can be used as a scan destination, similar to NullString. type NullBool struct { Bool bool Valid bool // Valid is true if Bool is not NULL } // Scan implements the Scanner interface. func (n *NullBool) Scan(value interface{}) error { if value == nil { n.Bool, n.Valid = false, false return nil } n.Valid = true return convertAssign(&n.Bool, value) } // Value implements the driver Valuer interface. func (n NullBool) Value() (driver.Value, error) { if !n.Valid { return nil, nil } return n.Bool, nil } // Scanner is an interface used by Scan. type Scanner interface { // Scan assigns a value from a database driver. // // The src value will be of one of the following types: // // int64 // float64 // bool // []byte // string // time.Time // nil - for NULL values // // An error should be returned if the value cannot be stored // without loss of information. Scan(src interface{}) error } // ErrNoRows is returned by Scan when QueryRow doesn't return a // row. In such a case, QueryRow returns a placeholder *Row value that // defers this error until a Scan. var ErrNoRows = errors.New("sql: no rows in result set") // DB is a database handle representing a pool of zero or more // underlying connections. It's safe for concurrent use by multiple // goroutines. // // The sql package creates and frees connections automatically; it // also maintains a free pool of idle connections. If the database has // a concept of per-connection state, such state can only be reliably // observed within a transaction. Once DB.Begin is called, the // returned Tx is bound to a single connection. Once Commit or // Rollback is called on the transaction, that transaction's // connection is returned to DB's idle connection pool. The pool size // can be controlled with SetMaxIdleConns. type DB struct { driver driver.Driver dsn string // numClosed is an atomic counter which represents a total number of // closed connections. Stmt.openStmt checks it before cleaning closed // connections in Stmt.css. numClosed uint64 mu sync.Mutex // protects following fields freeConn []*driverConn connRequests map[uint64]chan connRequest nextRequest uint64 // Next key to use in connRequests. numOpen int // number of opened and pending open connections // Used to signal the need for new connections // a goroutine running connectionOpener() reads on this chan and // maybeOpenNewConnections sends on the chan (one send per needed connection) // It is closed during db.Close(). The close tells the connectionOpener // goroutine to exit. openerCh chan struct{} closed bool dep map[finalCloser]depSet lastPut map[*driverConn]string // stacktrace of last conn's put; debug only maxIdle int // zero means defaultMaxIdleConns; negative means 0 maxOpen int // <= 0 means unlimited maxLifetime time.Duration // maximum amount of time a connection may be reused cleanerCh chan struct{} } // connReuseStrategy determines how (*DB).conn returns database connections. type connReuseStrategy uint8 const ( // alwaysNewConn forces a new connection to the database. alwaysNewConn connReuseStrategy = iota // cachedOrNewConn returns a cached connection, if available, else waits // for one to become available (if MaxOpenConns has been reached) or // creates a new database connection. cachedOrNewConn ) // driverConn wraps a driver.Conn with a mutex, to // be held during all calls into the Conn. (including any calls onto // interfaces returned via that Conn, such as calls on Tx, Stmt, // Result, Rows) type driverConn struct { db *DB createdAt time.Time sync.Mutex // guards following ci driver.Conn closed bool finalClosed bool // ci.Close has been called openStmt map[*driverStmt]bool // guarded by db.mu inUse bool onPut []func() // code (with db.mu held) run when conn is next returned dbmuClosed bool // same as closed, but guarded by db.mu, for removeClosedStmtLocked } func (dc *driverConn) releaseConn(err error) { dc.db.putConn(dc, err) } func (dc *driverConn) removeOpenStmt(ds *driverStmt) { dc.Lock() defer dc.Unlock() delete(dc.openStmt, ds) } func (dc *driverConn) expired(timeout time.Duration) bool { if timeout <= 0 { return false } return dc.createdAt.Add(timeout).Before(nowFunc()) } func (dc *driverConn) prepareLocked(ctx context.Context, query string) (*driverStmt, error) { si, err := ctxDriverPrepare(ctx, dc.ci, query) if err != nil { return nil, err } // Track each driverConn's open statements, so we can close them // before closing the conn. // // Wrap all driver.Stmt is *driverStmt to ensure they are only closed once. if dc.openStmt == nil { dc.openStmt = make(map[*driverStmt]bool) } ds := &driverStmt{Locker: dc, si: si} dc.openStmt[ds] = true return ds, nil } // the dc.db's Mutex is held. func (dc *driverConn) closeDBLocked() func() error { dc.Lock() defer dc.Unlock() if dc.closed { return func() error { return errors.New("sql: duplicate driverConn close") } } dc.closed = true return dc.db.removeDepLocked(dc, dc) } func (dc *driverConn) Close() error { dc.Lock() if dc.closed { dc.Unlock() return errors.New("sql: duplicate driverConn close") } dc.closed = true dc.Unlock() // not defer; removeDep finalClose calls may need to lock // And now updates that require holding dc.mu.Lock. dc.db.mu.Lock() dc.dbmuClosed = true fn := dc.db.removeDepLocked(dc, dc) dc.db.mu.Unlock() return fn() } func (dc *driverConn) finalClose() error { var err error // Each *driverStmt has a lock to the dc. Copy the list out of the dc // before calling close on each stmt. var openStmt []*driverStmt withLock(dc, func() { openStmt = make([]*driverStmt, 0, len(dc.openStmt)) for ds := range dc.openStmt { openStmt = append(openStmt, ds) } dc.openStmt = nil }) for _, ds := range openStmt { ds.Close() } withLock(dc, func() { dc.finalClosed = true err = dc.ci.Close() dc.ci = nil }) dc.db.mu.Lock() dc.db.numOpen-- dc.db.maybeOpenNewConnections() dc.db.mu.Unlock() atomic.AddUint64(&dc.db.numClosed, 1) return err } // driverStmt associates a driver.Stmt with the // *driverConn from which it came, so the driverConn's lock can be // held during calls. type driverStmt struct { sync.Locker // the *driverConn si driver.Stmt closed bool closeErr error // return value of previous Close call } // Close ensures dirver.Stmt is only closed once any always returns the same // result. func (ds *driverStmt) Close() error { ds.Lock() defer ds.Unlock() if ds.closed { return ds.closeErr } ds.closed = true ds.closeErr = ds.si.Close() return ds.closeErr } // depSet is a finalCloser's outstanding dependencies type depSet map[interface{}]bool // set of true bools // The finalCloser interface is used by (*DB).addDep and related // dependency reference counting. type finalCloser interface { // finalClose is called when the reference count of an object // goes to zero. (*DB).mu is not held while calling it. finalClose() error } // addDep notes that x now depends on dep, and x's finalClose won't be // called until all of x's dependencies are removed with removeDep. func (db *DB) addDep(x finalCloser, dep interface{}) { //println(fmt.Sprintf("addDep(%T %p, %T %p)", x, x, dep, dep)) db.mu.Lock() defer db.mu.Unlock() db.addDepLocked(x, dep) } func (db *DB) addDepLocked(x finalCloser, dep interface{}) { if db.dep == nil { db.dep = make(map[finalCloser]depSet) } xdep := db.dep[x] if xdep == nil { xdep = make(depSet) db.dep[x] = xdep } xdep[dep] = true } // removeDep notes that x no longer depends on dep. // If x still has dependencies, nil is returned. // If x no longer has any dependencies, its finalClose method will be // called and its error value will be returned. func (db *DB) removeDep(x finalCloser, dep interface{}) error { db.mu.Lock() fn := db.removeDepLocked(x, dep) db.mu.Unlock() return fn() } func (db *DB) removeDepLocked(x finalCloser, dep interface{}) func() error { //println(fmt.Sprintf("removeDep(%T %p, %T %p)", x, x, dep, dep)) xdep, ok := db.dep[x] if !ok { panic(fmt.Sprintf("unpaired removeDep: no deps for %T", x)) } l0 := len(xdep) delete(xdep, dep) switch len(xdep) { case l0: // Nothing removed. Shouldn't happen. panic(fmt.Sprintf("unpaired removeDep: no %T dep on %T", dep, x)) case 0: // No more dependencies. delete(db.dep, x) return x.finalClose default: // Dependencies remain. return func() error { return nil } } } // This is the size of the connectionOpener request chan (DB.openerCh). // This value should be larger than the maximum typical value // used for db.maxOpen. If maxOpen is significantly larger than // connectionRequestQueueSize then it is possible for ALL calls into the *DB // to block until the connectionOpener can satisfy the backlog of requests. var connectionRequestQueueSize = 1000000 // Open opens a database specified by its database driver name and a // driver-specific data source name, usually consisting of at least a // database name and connection information. // // Most users will open a database via a driver-specific connection // helper function that returns a *DB. No database drivers are included // in the Go standard library. See https://golang.org/s/sqldrivers for // a list of third-party drivers. // // Open may just validate its arguments without creating a connection // to the database. To verify that the data source name is valid, call // Ping. // // The returned DB is safe for concurrent use by multiple goroutines // and maintains its own pool of idle connections. Thus, the Open // function should be called just once. It is rarely necessary to // close a DB. func Open(driverName, dataSourceName string) (*DB, error) { driversMu.RLock() driveri, ok := drivers[driverName] driversMu.RUnlock() if !ok { return nil, fmt.Errorf("sql: unknown driver %q (forgotten import?)", driverName) } db := &DB{ driver: driveri, dsn: dataSourceName, openerCh: make(chan struct{}, connectionRequestQueueSize), lastPut: make(map[*driverConn]string), connRequests: make(map[uint64]chan connRequest), } go db.connectionOpener() return db, nil } // PingContext verifies a connection to the database is still alive, // establishing a connection if necessary. func (db *DB) PingContext(ctx context.Context) error { var dc *driverConn var err error for i := 0; i < maxBadConnRetries; i++ { dc, err = db.conn(ctx, cachedOrNewConn) if err != driver.ErrBadConn { break } } if err == driver.ErrBadConn { dc, err = db.conn(ctx, alwaysNewConn) } if err != nil { return err } if pinger, ok := dc.ci.(driver.Pinger); ok { err = pinger.Ping(ctx) } db.putConn(dc, err) return err } // Ping verifies a connection to the database is still alive, // establishing a connection if necessary. func (db *DB) Ping() error { return db.PingContext(context.Background()) } // Close closes the database, releasing any open resources. // // It is rare to Close a DB, as the DB handle is meant to be // long-lived and shared between many goroutines. func (db *DB) Close() error { db.mu.Lock() if db.closed { // Make DB.Close idempotent db.mu.Unlock() return nil } close(db.openerCh) if db.cleanerCh != nil { close(db.cleanerCh) } var err error fns := make([]func() error, 0, len(db.freeConn)) for _, dc := range db.freeConn { fns = append(fns, dc.closeDBLocked()) } db.freeConn = nil db.closed = true for _, req := range db.connRequests { close(req) } db.mu.Unlock() for _, fn := range fns { err1 := fn() if err1 != nil { err = err1 } } return err } const defaultMaxIdleConns = 2 func (db *DB) maxIdleConnsLocked() int { n := db.maxIdle switch { case n == 0: // TODO(bradfitz): ask driver, if supported, for its default preference return defaultMaxIdleConns case n < 0: return 0 default: return n } } // SetMaxIdleConns sets the maximum number of connections in the idle // connection pool. // // If MaxOpenConns is greater than 0 but less than the new MaxIdleConns // then the new MaxIdleConns will be reduced to match the MaxOpenConns limit // // If n <= 0, no idle connections are retained. func (db *DB) SetMaxIdleConns(n int) { db.mu.Lock() if n > 0 { db.maxIdle = n } else { // No idle connections. db.maxIdle = -1 } // Make sure maxIdle doesn't exceed maxOpen if db.maxOpen > 0 && db.maxIdleConnsLocked() > db.maxOpen { db.maxIdle = db.maxOpen } var closing []*driverConn idleCount := len(db.freeConn) maxIdle := db.maxIdleConnsLocked() if idleCount > maxIdle { closing = db.freeConn[maxIdle:] db.freeConn = db.freeConn[:maxIdle] } db.mu.Unlock() for _, c := range closing { c.Close() } } // SetMaxOpenConns sets the maximum number of open connections to the database. // // If MaxIdleConns is greater than 0 and the new MaxOpenConns is less than // MaxIdleConns, then MaxIdleConns will be reduced to match the new // MaxOpenConns limit // // If n <= 0, then there is no limit on the number of open connections. // The default is 0 (unlimited). func (db *DB) SetMaxOpenConns(n int) { db.mu.Lock() db.maxOpen = n if n < 0 { db.maxOpen = 0 } syncMaxIdle := db.maxOpen > 0 && db.maxIdleConnsLocked() > db.maxOpen db.mu.Unlock() if syncMaxIdle { db.SetMaxIdleConns(n) } } // SetConnMaxLifetime sets the maximum amount of time a connection may be reused. // // Expired connections may be closed lazily before reuse. // // If d <= 0, connections are reused forever. func (db *DB) SetConnMaxLifetime(d time.Duration) { if d < 0 { d = 0 } db.mu.Lock() // wake cleaner up when lifetime is shortened. if d > 0 && d < db.maxLifetime && db.cleanerCh != nil { select { case db.cleanerCh <- struct{}{}: default: } } db.maxLifetime = d db.startCleanerLocked() db.mu.Unlock() } // startCleanerLocked starts connectionCleaner if needed. func (db *DB) startCleanerLocked() { if db.maxLifetime > 0 && db.numOpen > 0 && db.cleanerCh == nil { db.cleanerCh = make(chan struct{}, 1) go db.connectionCleaner(db.maxLifetime) } } func (db *DB) connectionCleaner(d time.Duration) { const minInterval = time.Second if d < minInterval { d = minInterval } t := time.NewTimer(d) for { select { case <-t.C: case <-db.cleanerCh: // maxLifetime was changed or db was closed. } db.mu.Lock() d = db.maxLifetime if db.closed || db.numOpen == 0 || d <= 0 { db.cleanerCh = nil db.mu.Unlock() return } expiredSince := nowFunc().Add(-d) var closing []*driverConn for i := 0; i < len(db.freeConn); i++ { c := db.freeConn[i] if c.createdAt.Before(expiredSince) { closing = append(closing, c) last := len(db.freeConn) - 1 db.freeConn[i] = db.freeConn[last] db.freeConn[last] = nil db.freeConn = db.freeConn[:last] i-- } } db.mu.Unlock() for _, c := range closing { c.Close() } if d < minInterval { d = minInterval } t.Reset(d) } } // DBStats contains database statistics. type DBStats struct { // OpenConnections is the number of open connections to the database. OpenConnections int } // Stats returns database statistics. func (db *DB) Stats() DBStats { db.mu.Lock() stats := DBStats{ OpenConnections: db.numOpen, } db.mu.Unlock() return stats } // Assumes db.mu is locked. // If there are connRequests and the connection limit hasn't been reached, // then tell the connectionOpener to open new connections. func (db *DB) maybeOpenNewConnections() { numRequests := len(db.connRequests) if db.maxOpen > 0 { numCanOpen := db.maxOpen - db.numOpen if numRequests > numCanOpen { numRequests = numCanOpen } } for numRequests > 0 { db.numOpen++ // optimistically numRequests-- if db.closed { return } db.openerCh <- struct{}{} } } // Runs in a separate goroutine, opens new connections when requested. func (db *DB) connectionOpener() { for range db.openerCh { db.openNewConnection() } } // Open one new connection func (db *DB) openNewConnection() { // maybeOpenNewConnctions has already executed db.numOpen++ before it sent // on db.openerCh. This function must execute db.numOpen-- if the // connection fails or is closed before returning. ci, err := db.driver.Open(db.dsn) db.mu.Lock() defer db.mu.Unlock() if db.closed { if err == nil { ci.Close() } db.numOpen-- return } if err != nil { db.numOpen-- db.putConnDBLocked(nil, err) db.maybeOpenNewConnections() return } dc := &driverConn{ db: db, createdAt: nowFunc(), ci: ci, } if db.putConnDBLocked(dc, err) { db.addDepLocked(dc, dc) } else { db.numOpen-- ci.Close() } } // connRequest represents one request for a new connection // When there are no idle connections available, DB.conn will create // a new connRequest and put it on the db.connRequests list. type connRequest struct { conn *driverConn err error } var errDBClosed = errors.New("sql: database is closed") // nextRequestKeyLocked returns the next connection request key. // It is assumed that nextRequest will not overflow. func (db *DB) nextRequestKeyLocked() uint64 { next := db.nextRequest db.nextRequest++ return next } // conn returns a newly-opened or cached *driverConn. func (db *DB) conn(ctx context.Context, strategy connReuseStrategy) (*driverConn, error) { db.mu.Lock() if db.closed { db.mu.Unlock() return nil, errDBClosed } // Check if the context is expired. select { default: case <-ctx.Done(): db.mu.Unlock() return nil, ctx.Err() } lifetime := db.maxLifetime // Prefer a free connection, if possible. numFree := len(db.freeConn) if strategy == cachedOrNewConn && numFree > 0 { conn := db.freeConn[0] copy(db.freeConn, db.freeConn[1:]) db.freeConn = db.freeConn[:numFree-1] conn.inUse = true db.mu.Unlock() if conn.expired(lifetime) { conn.Close() return nil, driver.ErrBadConn } return conn, nil } // Out of free connections or we were asked not to use one. If we're not // allowed to open any more connections, make a request and wait. if db.maxOpen > 0 && db.numOpen >= db.maxOpen { // Make the connRequest channel. It's buffered so that the // connectionOpener doesn't block while waiting for the req to be read. req := make(chan connRequest, 1) reqKey := db.nextRequestKeyLocked() db.connRequests[reqKey] = req db.mu.Unlock() // Timeout the connection request with the context. select { case <-ctx.Done(): // Remove the connection request and ensure no value has been sent // on it after removing. db.mu.Lock() delete(db.connRequests, reqKey) db.mu.Unlock() select { default: case ret, ok := <-req: if ok { db.putConn(ret.conn, ret.err) } } return nil, ctx.Err() case ret, ok := <-req: if !ok { return nil, errDBClosed } if ret.err == nil && ret.conn.expired(lifetime) { ret.conn.Close() return nil, driver.ErrBadConn } return ret.conn, ret.err } } db.numOpen++ // optimistically db.mu.Unlock() ci, err := db.driver.Open(db.dsn) if err != nil { db.mu.Lock() db.numOpen-- // correct for earlier optimism db.maybeOpenNewConnections() db.mu.Unlock() return nil, err } db.mu.Lock() dc := &driverConn{ db: db, createdAt: nowFunc(), ci: ci, } db.addDepLocked(dc, dc) dc.inUse = true db.mu.Unlock() return dc, nil } // putConnHook is a hook for testing. var putConnHook func(*DB, *driverConn) // noteUnusedDriverStatement notes that ds is no longer used and should // be closed whenever possible (when c is next not in use), unless c is // already closed. func (db *DB) noteUnusedDriverStatement(c *driverConn, ds *driverStmt) { db.mu.Lock() defer db.mu.Unlock() if c.inUse { c.onPut = append(c.onPut, func() { ds.Close() }) } else { c.Lock() fc := c.finalClosed c.Unlock() if !fc { ds.Close() } } } // debugGetPut determines whether getConn & putConn calls' stack traces // are returned for more verbose crashes. const debugGetPut = false // putConn adds a connection to the db's free pool. // err is optionally the last error that occurred on this connection. func (db *DB) putConn(dc *driverConn, err error) { db.mu.Lock() if !dc.inUse { if debugGetPut { fmt.Printf("putConn(%v) DUPLICATE was: %s\n\nPREVIOUS was: %s", dc, stack(), db.lastPut[dc]) } panic("sql: connection returned that was never out") } if debugGetPut { db.lastPut[dc] = stack() } dc.inUse = false for _, fn := range dc.onPut { fn() } dc.onPut = nil if err == driver.ErrBadConn { // Don't reuse bad connections. // Since the conn is considered bad and is being discarded, treat it // as closed. Don't decrement the open count here, finalClose will // take care of that. db.maybeOpenNewConnections() db.mu.Unlock() dc.Close() return } if putConnHook != nil { putConnHook(db, dc) } added := db.putConnDBLocked(dc, nil) db.mu.Unlock() if !added { dc.Close() } } // Satisfy a connRequest or put the driverConn in the idle pool and return true // or return false. // putConnDBLocked will satisfy a connRequest if there is one, or it will // return the *driverConn to the freeConn list if err == nil and the idle // connection limit will not be exceeded. // If err != nil, the value of dc is ignored. // If err == nil, then dc must not equal nil. // If a connRequest was fulfilled or the *driverConn was placed in the // freeConn list, then true is returned, otherwise false is returned. func (db *DB) putConnDBLocked(dc *driverConn, err error) bool { if db.closed { return false } if db.maxOpen > 0 && db.numOpen > db.maxOpen { return false } if c := len(db.connRequests); c > 0 { var req chan connRequest var reqKey uint64 for reqKey, req = range db.connRequests { break } delete(db.connRequests, reqKey) // Remove from pending requests. if err == nil { dc.inUse = true } req <- connRequest{ conn: dc, err: err, } return true } else if err == nil && !db.closed && db.maxIdleConnsLocked() > len(db.freeConn) { db.freeConn = append(db.freeConn, dc) db.startCleanerLocked() return true } return false } // maxBadConnRetries is the number of maximum retries if the driver returns // driver.ErrBadConn to signal a broken connection before forcing a new // connection to be opened. const maxBadConnRetries = 2 // PrepareContext 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 provided context is used for the preparation of the statement, not for the // execution of the statement. func (db *DB) PrepareContext(ctx context.Context, query string) (*Stmt, error) { var stmt *Stmt var err error for i := 0; i < maxBadConnRetries; i++ { stmt, err = db.prepare(ctx, query, cachedOrNewConn) if err != driver.ErrBadConn { break } } if err == driver.ErrBadConn { return db.prepare(ctx, query, alwaysNewConn) } return stmt, err } // 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. func (db *DB) Prepare(query string) (*Stmt, error) { return db.PrepareContext(context.Background(), query) } func (db *DB) prepare(ctx context.Context, query string, strategy connReuseStrategy) (*Stmt, error) { // TODO: check if db.driver supports an optional // driver.Preparer interface and call that instead, if so, // otherwise we make a prepared statement that's bound // to a connection, and to execute this prepared statement // we either need to use this connection (if it's free), else // get a new connection + re-prepare + execute on that one. dc, err := db.conn(ctx, strategy) if err != nil { return nil, err } var ds *driverStmt withLock(dc, func() { ds, err = dc.prepareLocked(ctx, query) }) if err != nil { db.putConn(dc, err) return nil, err } stmt := &Stmt{ db: db, query: query, css: []connStmt{{dc, ds}}, lastNumClosed: atomic.LoadUint64(&db.numClosed), } db.addDep(stmt, stmt) db.putConn(dc, nil) return stmt, nil } // ExecContext executes a query without returning any rows. // The args are for any placeholder parameters in the query. func (db *DB) ExecContext(ctx context.Context, query string, args ...interface{}) (Result, error) { var res Result var err error for i := 0; i < maxBadConnRetries; i++ { res, err = db.exec(ctx, query, args, cachedOrNewConn) if err != driver.ErrBadConn { break } } if err == driver.ErrBadConn { return db.exec(ctx, query, args, alwaysNewConn) } return res, err } // Exec executes a query without returning any rows. // The args are for any placeholder parameters in the query. func (db *DB) Exec(query string, args ...interface{}) (Result, error) { return db.ExecContext(context.Background(), query, args...) } func (db *DB) exec(ctx context.Context, query string, args []interface{}, strategy connReuseStrategy) (res Result, err error) { dc, err := db.conn(ctx, strategy) if err != nil { return nil, err } defer func() { db.putConn(dc, err) }() if execer, ok := dc.ci.(driver.Execer); ok { var dargs []driver.NamedValue dargs, err = driverArgs(nil, args) if err != nil { return nil, err } var resi driver.Result withLock(dc, func() { resi, err = ctxDriverExec(ctx, execer, query, dargs) }) if err != driver.ErrSkip { if err != nil { return nil, err } return driverResult{dc, resi}, nil } } var si driver.Stmt withLock(dc, func() { si, err = ctxDriverPrepare(ctx, dc.ci, query) }) if err != nil { return nil, err } ds := &driverStmt{Locker: dc, si: si} defer ds.Close() return resultFromStatement(ctx, ds, args...) } // QueryContext executes a query that returns rows, typically a SELECT. // The args are for any placeholder parameters in the query. func (db *DB) QueryContext(ctx context.Context, query string, args ...interface{}) (*Rows, error) { var rows *Rows var err error for i := 0; i < maxBadConnRetries; i++ { rows, err = db.query(ctx, query, args, cachedOrNewConn) if err != driver.ErrBadConn { break } } if err == driver.ErrBadConn { return db.query(ctx, query, args, alwaysNewConn) } return rows, err } // Query executes a query that returns rows, typically a SELECT. // The args are for any placeholder parameters in the query. func (db *DB) Query(query string, args ...interface{}) (*Rows, error) { return db.QueryContext(context.Background(), query, args...) } func (db *DB) query(ctx context.Context, query string, args []interface{}, strategy connReuseStrategy) (*Rows, error) { ci, err := db.conn(ctx, strategy) if err != nil { return nil, err } return db.queryConn(ctx, ci, ci.releaseConn, query, args) } // queryConn executes a query on the given connection. // The connection gets released by the releaseConn function. func (db *DB) queryConn(ctx context.Context, dc *driverConn, releaseConn func(error), query string, args []interface{}) (*Rows, error) { if queryer, ok := dc.ci.(driver.Queryer); ok { dargs, err := driverArgs(nil, args) if err != nil { releaseConn(err) return nil, err } var rowsi driver.Rows withLock(dc, func() { rowsi, err = ctxDriverQuery(ctx, queryer, query, dargs) }) if err != driver.ErrSkip { if err != nil { releaseConn(err) return nil, err } // Note: ownership of dc passes to the *Rows, to be freed // with releaseConn. rows := &Rows{ dc: dc, releaseConn: releaseConn, rowsi: rowsi, } rows.initContextClose(ctx) return rows, nil } } var si driver.Stmt var err error withLock(dc, func() { si, err = ctxDriverPrepare(ctx, dc.ci, query) }) if err != nil { releaseConn(err) return nil, err } ds := &driverStmt{Locker: dc, si: si} rowsi, err := rowsiFromStatement(ctx, ds, args...) if err != nil { ds.Close() releaseConn(err) return nil, err } // Note: ownership of ci passes to the *Rows, to be freed // with releaseConn. rows := &Rows{ dc: dc, releaseConn: releaseConn, rowsi: rowsi, closeStmt: ds, } rows.initContextClose(ctx) return rows, nil } // QueryRowContext executes a query that is expected to return at most one row. // QueryRowContext always returns a non-nil value. Errors are deferred until // Row's Scan method is called. func (db *DB) QueryRowContext(ctx context.Context, query string, args ...interface{}) *Row { rows, err := db.QueryContext(ctx, query, args...) return &Row{rows: rows, err: err} } // QueryRow executes a query that is expected to return at most one row. // QueryRow always returns a non-nil value. Errors are deferred until // Row's Scan method is called. func (db *DB) QueryRow(query string, args ...interface{}) *Row { return db.QueryRowContext(context.Background(), query, args...) } // BeginTx starts a transaction. // // The provided context is used until the transaction is committed or rolled back. // If the context is canceled, the sql package will roll back // the transaction. Tx.Commit will return an error if the context provided to // BeginTx is canceled. // // The provided TxOptions is optional and may be nil if defaults should be used. // If a non-default isolation level is used that the driver doesn't support, // an error will be returned. func (db *DB) BeginTx(ctx context.Context, opts *TxOptions) (*Tx, error) { var tx *Tx var err error for i := 0; i < maxBadConnRetries; i++ { tx, err = db.begin(ctx, opts, cachedOrNewConn) if err != driver.ErrBadConn { break } } if err == driver.ErrBadConn { return db.begin(ctx, opts, alwaysNewConn) } return tx, err } // Begin starts a transaction. The default isolation level is dependent on // the driver. func (db *DB) Begin() (*Tx, error) { return db.BeginTx(context.Background(), nil) } func (db *DB) begin(ctx context.Context, opts *TxOptions, strategy connReuseStrategy) (tx *Tx, err error) { dc, err := db.conn(ctx, strategy) if err != nil { return nil, err } var txi driver.Tx withLock(dc, func() { txi, err = ctxDriverBegin(ctx, opts, dc.ci) }) if err != nil { db.putConn(dc, err) return nil, err } // Schedule the transaction to rollback when the context is cancelled. // The cancel function in Tx will be called after done is set to true. ctx, cancel := context.WithCancel(ctx) tx = &Tx{ db: db, dc: dc, txi: txi, cancel: cancel, ctx: ctx, } go tx.awaitDone() return tx, nil } // Driver returns the database's underlying driver. func (db *DB) Driver() driver.Driver { return db.driver } // Tx is an in-progress database transaction. // // A transaction must end with a call to Commit or Rollback. // // After a call to Commit or Rollback, all operations on the // transaction fail with ErrTxDone. // // The statements prepared for a transaction by calling // the transaction's Prepare or Stmt methods are closed // by the call to Commit or Rollback. type Tx struct { db *DB // closemu prevents the transaction from closing while there // is an active query. It is held for read during queries // and exclusively during close. closemu sync.RWMutex // dc is owned exclusively until Commit or Rollback, at which point // it's returned with putConn. dc *driverConn txi driver.Tx // done transitions from 0 to 1 exactly once, on Commit // or Rollback. once done, all operations fail with // ErrTxDone. // Use atomic operations on value when checking value. done int32 // All Stmts prepared for this transaction. These will be closed after the // transaction has been committed or rolled back. stmts struct { sync.Mutex v []*Stmt } // cancel is called after done transitions from false to true. cancel func() // ctx lives for the life of the transaction. ctx context.Context } // awaitDone blocks until the context in Tx is canceled and rolls back // the transaction if it's not already done. func (tx *Tx) awaitDone() { // Wait for either the transaction to be committed or rolled // back, or for the associated context to be closed. <-tx.ctx.Done() // Discard and close the connection used to ensure the // transaction is closed and the resources are released. This // rollback does nothing if the transaction has already been // committed or rolled back. tx.rollback(true) } func (tx *Tx) isDone() bool { return atomic.LoadInt32(&tx.done) != 0 } // ErrTxDone is returned by any operation that is performed on a transaction // that has already been committed or rolled back. var ErrTxDone = errors.New("sql: Transaction has already been committed or rolled back") // close returns the connection to the pool and // must only be called by Tx.rollback or Tx.Commit. func (tx *Tx) close(err error) { tx.closemu.Lock() defer tx.closemu.Unlock() tx.db.putConn(tx.dc, err) tx.cancel() tx.dc = nil tx.txi = nil } // hookTxGrabConn specifies an optional hook to be called on // a successful call to (*Tx).grabConn. For tests. var hookTxGrabConn func() func (tx *Tx) grabConn(ctx context.Context) (*driverConn, error) { select { default: case <-ctx.Done(): return nil, ctx.Err() } if tx.isDone() { return nil, ErrTxDone } if hookTxGrabConn != nil { // test hook hookTxGrabConn() } return tx.dc, nil } // Closes all Stmts prepared for this transaction. func (tx *Tx) closePrepared() { tx.stmts.Lock() defer tx.stmts.Unlock() for _, stmt := range tx.stmts.v { stmt.Close() } } // Commit commits the transaction. func (tx *Tx) Commit() error { if !atomic.CompareAndSwapInt32(&tx.done, 0, 1) { return ErrTxDone } select { default: case <-tx.ctx.Done(): return tx.ctx.Err() } var err error withLock(tx.dc, func() { err = tx.txi.Commit() }) if err != driver.ErrBadConn { tx.closePrepared() } tx.close(err) return err } // rollback aborts the transaction and optionally forces the pool to discard // the connection. func (tx *Tx) rollback(discardConn bool) error { if !atomic.CompareAndSwapInt32(&tx.done, 0, 1) { return ErrTxDone } var err error withLock(tx.dc, func() { err = tx.txi.Rollback() }) if err != driver.ErrBadConn { tx.closePrepared() } if discardConn { err = driver.ErrBadConn } tx.close(err) return err } // Rollback aborts the transaction. func (tx *Tx) Rollback() error { return tx.rollback(false) } // Prepare creates a prepared statement for use within a transaction. // // The returned statement operates within the transaction and will be closed // when the transaction has been committed or rolled back. // // To use an existing prepared statement on this transaction, see Tx.Stmt. // // The provided context will be used for the preparation of the context, not // for the execution of the returned statement. The returned statement // will run in the transaction context. func (tx *Tx) PrepareContext(ctx context.Context, query string) (*Stmt, error) { tx.closemu.RLock() defer tx.closemu.RUnlock() // TODO(bradfitz): We could be more efficient here and either // provide a method to take an existing Stmt (created on // perhaps a different Conn), and re-create it on this Conn if // necessary. Or, better: keep a map in DB of query string to // Stmts, and have Stmt.Execute do the right thing and // re-prepare if the Conn in use doesn't have that prepared // statement. But we'll want to avoid caching the statement // in the case where we only call conn.Prepare implicitly // (such as in db.Exec or tx.Exec), but the caller package // can't be holding a reference to the returned statement. // Perhaps just looking at the reference count (by noting // Stmt.Close) would be enough. We might also want a finalizer // on Stmt to drop the reference count. dc, err := tx.grabConn(ctx) if err != nil { return nil, err } var si driver.Stmt withLock(dc, func() { si, err = ctxDriverPrepare(ctx, dc.ci, query) }) if err != nil { return nil, err } stmt := &Stmt{ db: tx.db, tx: tx, txds: &driverStmt{ Locker: dc, si: si, }, query: query, } tx.stmts.Lock() tx.stmts.v = append(tx.stmts.v, stmt) tx.stmts.Unlock() return stmt, nil } // Prepare creates a prepared statement for use within a transaction. // // The returned statement operates within the transaction and can no longer // be used once the transaction has been committed or rolled back. // // To use an existing prepared statement on this transaction, see Tx.Stmt. func (tx *Tx) Prepare(query string) (*Stmt, error) { return tx.PrepareContext(context.Background(), query) } // StmtContext returns a transaction-specific prepared statement from // an existing statement. // // Example: // updateMoney, err := db.Prepare("UPDATE balance SET money=money+? WHERE id=?") // ... // tx, err := db.Begin() // ... // res, err := tx.StmtContext(ctx, updateMoney).Exec(123.45, 98293203) // // The returned statement operates within the transaction and will be closed // when the transaction has been committed or rolled back. func (tx *Tx) StmtContext(ctx context.Context, stmt *Stmt) *Stmt { tx.closemu.RLock() defer tx.closemu.RUnlock() // TODO(bradfitz): optimize this. Currently this re-prepares // each time. This is fine for now to illustrate the API but // we should really cache already-prepared statements // per-Conn. See also the big comment in Tx.Prepare. if tx.db != stmt.db { return &Stmt{stickyErr: errors.New("sql: Tx.Stmt: statement from different database used")} } dc, err := tx.grabConn(ctx) if err != nil { return &Stmt{stickyErr: err} } var si driver.Stmt withLock(dc, func() { si, err = ctxDriverPrepare(ctx, dc.ci, stmt.query) }) txs := &Stmt{ db: tx.db, tx: tx, txds: &driverStmt{ Locker: dc, si: si, }, query: stmt.query, stickyErr: err, } tx.stmts.Lock() tx.stmts.v = append(tx.stmts.v, txs) tx.stmts.Unlock() return txs } // Stmt returns a transaction-specific prepared statement from // an existing statement. // // Example: // updateMoney, err := db.Prepare("UPDATE balance SET money=money+? WHERE id=?") // ... // tx, err := db.Begin() // ... // res, err := tx.Stmt(updateMoney).Exec(123.45, 98293203) // // The returned statement operates within the transaction and will be closed // when the transaction has been committed or rolled back. func (tx *Tx) Stmt(stmt *Stmt) *Stmt { return tx.StmtContext(context.Background(), stmt) } // ExecContext executes a query that doesn't return rows. // For example: an INSERT and UPDATE. func (tx *Tx) ExecContext(ctx context.Context, query string, args ...interface{}) (Result, error) { tx.closemu.RLock() defer tx.closemu.RUnlock() dc, err := tx.grabConn(ctx) if err != nil { return nil, err } if execer, ok := dc.ci.(driver.Execer); ok { dargs, err := driverArgs(nil, args) if err != nil { return nil, err } var resi driver.Result withLock(dc, func() { resi, err = ctxDriverExec(ctx, execer, query, dargs) }) if err == nil { return driverResult{dc, resi}, nil } if err != driver.ErrSkip { return nil, err } } var si driver.Stmt withLock(dc, func() { si, err = ctxDriverPrepare(ctx, dc.ci, query) }) if err != nil { return nil, err } ds := &driverStmt{Locker: dc, si: si} defer ds.Close() return resultFromStatement(ctx, ds, args...) } // Exec executes a query that doesn't return rows. // For example: an INSERT and UPDATE. func (tx *Tx) Exec(query string, args ...interface{}) (Result, error) { return tx.ExecContext(context.Background(), query, args...) } // QueryContext executes a query that returns rows, typically a SELECT. func (tx *Tx) QueryContext(ctx context.Context, query string, args ...interface{}) (*Rows, error) { tx.closemu.RLock() defer tx.closemu.RUnlock() dc, err := tx.grabConn(ctx) if err != nil { return nil, err } releaseConn := func(error) {} return tx.db.queryConn(ctx, dc, releaseConn, query, args) } // Query executes a query that returns rows, typically a SELECT. func (tx *Tx) Query(query string, args ...interface{}) (*Rows, error) { return tx.QueryContext(context.Background(), query, args...) } // QueryRowContext executes a query that is expected to return at most one row. // QueryRowContext always returns a non-nil value. Errors are deferred until // Row's Scan method is called. func (tx *Tx) QueryRowContext(ctx context.Context, query string, args ...interface{}) *Row { rows, err := tx.QueryContext(ctx, query, args...) return &Row{rows: rows, err: err} } // QueryRow executes a query that is expected to return at most one row. // QueryRow always returns a non-nil value. Errors are deferred until // Row's Scan method is called. func (tx *Tx) QueryRow(query string, args ...interface{}) *Row { return tx.QueryRowContext(context.Background(), query, args...) } // connStmt is a prepared statement on a particular connection. type connStmt struct { dc *driverConn ds *driverStmt } // Stmt is a prepared statement. // A Stmt is safe for concurrent use by multiple goroutines. type Stmt struct { // Immutable: db *DB // where we came from query string // that created the Stmt stickyErr error // if non-nil, this error is returned for all operations closemu sync.RWMutex // held exclusively during close, for read otherwise. // If in a transaction, else both nil: tx *Tx txds *driverStmt mu sync.Mutex // protects the rest of the fields closed bool // css is a list of underlying driver statement interfaces // that are valid on particular connections. This is only // used if tx == nil and one is found that has idle // connections. If tx != nil, txsi is always used. css []connStmt // lastNumClosed is copied from db.numClosed when Stmt is created // without tx and closed connections in css are removed. lastNumClosed uint64 } // ExecContext executes a prepared statement with the given arguments and // returns a Result summarizing the effect of the statement. func (s *Stmt) ExecContext(ctx context.Context, args ...interface{}) (Result, error) { s.closemu.RLock() defer s.closemu.RUnlock() var res Result for i := 0; i < maxBadConnRetries; i++ { _, releaseConn, ds, err := s.connStmt(ctx) if err != nil { if err == driver.ErrBadConn { continue } return nil, err } res, err = resultFromStatement(ctx, ds, args...) releaseConn(err) if err != driver.ErrBadConn { return res, err } } return nil, driver.ErrBadConn } // Exec executes a prepared statement with the given arguments and // returns a Result summarizing the effect of the statement. func (s *Stmt) Exec(args ...interface{}) (Result, error) { return s.ExecContext(context.Background(), args...) } func driverNumInput(ds *driverStmt) int { ds.Lock() defer ds.Unlock() // in case NumInput panics return ds.si.NumInput() } func resultFromStatement(ctx context.Context, ds *driverStmt, args ...interface{}) (Result, error) { want := driverNumInput(ds) // -1 means the driver doesn't know how to count the number of // placeholders, so we won't sanity check input here and instead let the // driver deal with errors. if want != -1 && len(args) != want { return nil, fmt.Errorf("sql: expected %d arguments, got %d", want, len(args)) } dargs, err := driverArgs(ds, args) if err != nil { return nil, err } ds.Lock() defer ds.Unlock() resi, err := ctxDriverStmtExec(ctx, ds.si, dargs) if err != nil { return nil, err } return driverResult{ds.Locker, resi}, nil } // removeClosedStmtLocked removes closed conns in s.css. // // To avoid lock contention on DB.mu, we do it only when // s.db.numClosed - s.lastNum is large enough. func (s *Stmt) removeClosedStmtLocked() { t := len(s.css)/2 + 1 if t > 10 { t = 10 } dbClosed := atomic.LoadUint64(&s.db.numClosed) if dbClosed-s.lastNumClosed < uint64(t) { return } s.db.mu.Lock() for i := 0; i < len(s.css); i++ { if s.css[i].dc.dbmuClosed { s.css[i] = s.css[len(s.css)-1] s.css = s.css[:len(s.css)-1] i-- } } s.db.mu.Unlock() s.lastNumClosed = dbClosed } // connStmt returns a free driver connection on which to execute the // statement, a function to call to release the connection, and a // statement bound to that connection. func (s *Stmt) connStmt(ctx context.Context) (ci *driverConn, releaseConn func(error), ds *driverStmt, err error) { if err = s.stickyErr; err != nil { return } s.mu.Lock() if s.closed { s.mu.Unlock() err = errors.New("sql: statement is closed") return } // In a transaction, we always use the connection that the // transaction was created on. if s.tx != nil { s.mu.Unlock() ci, err = s.tx.grabConn(ctx) // blocks, waiting for the connection. if err != nil { return } releaseConn = func(error) {} return ci, releaseConn, s.txds, nil } s.removeClosedStmtLocked() s.mu.Unlock() dc, err := s.db.conn(ctx, cachedOrNewConn) if err != nil { return nil, nil, nil, err } s.mu.Lock() for _, v := range s.css { if v.dc == dc { s.mu.Unlock() return dc, dc.releaseConn, v.ds, nil } } s.mu.Unlock() // No luck; we need to prepare the statement on this connection withLock(dc, func() { ds, err = dc.prepareLocked(ctx, s.query) }) if err != nil { s.db.putConn(dc, err) return nil, nil, nil, err } s.mu.Lock() cs := connStmt{dc, ds} s.css = append(s.css, cs) s.mu.Unlock() return dc, dc.releaseConn, ds, nil } // QueryContext executes a prepared query statement with the given arguments // and returns the query results as a *Rows. func (s *Stmt) QueryContext(ctx context.Context, args ...interface{}) (*Rows, error) { s.closemu.RLock() defer s.closemu.RUnlock() var rowsi driver.Rows for i := 0; i < maxBadConnRetries; i++ { dc, releaseConn, ds, err := s.connStmt(ctx) if err != nil { if err == driver.ErrBadConn { continue } return nil, err } rowsi, err = rowsiFromStatement(ctx, ds, args...) if err == nil { // Note: ownership of ci passes to the *Rows, to be freed // with releaseConn. rows := &Rows{ dc: dc, rowsi: rowsi, // releaseConn set below } s.db.addDep(s, rows) rows.releaseConn = func(err error) { releaseConn(err) s.db.removeDep(s, rows) } rows.initContextClose(ctx) return rows, nil } releaseConn(err) if err != driver.ErrBadConn { return nil, err } } return nil, driver.ErrBadConn } // Query executes a prepared query statement with the given arguments // and returns the query results as a *Rows. func (s *Stmt) Query(args ...interface{}) (*Rows, error) { return s.QueryContext(context.Background(), args...) } func rowsiFromStatement(ctx context.Context, ds *driverStmt, args ...interface{}) (driver.Rows, error) { var want int withLock(ds, func() { want = ds.si.NumInput() }) // -1 means the driver doesn't know how to count the number of // placeholders, so we won't sanity check input here and instead let the // driver deal with errors. if want != -1 && len(args) != want { return nil, fmt.Errorf("sql: statement expects %d inputs; got %d", want, len(args)) } dargs, err := driverArgs(ds, args) if err != nil { return nil, err } ds.Lock() defer ds.Unlock() rowsi, err := ctxDriverStmtQuery(ctx, ds.si, dargs) if err != nil { return nil, err } return rowsi, nil } // QueryRowContext executes a prepared query statement with the given arguments. // If an error occurs during the execution of the statement, that error will // be returned by a call to Scan on the returned *Row, which is always non-nil. // If the query selects no rows, the *Row's Scan will return ErrNoRows. // Otherwise, the *Row's Scan scans the first selected row and discards // the rest. // // Example usage: // // var name string // err := nameByUseridStmt.QueryRowContext(ctx, id).Scan(&name) func (s *Stmt) QueryRowContext(ctx context.Context, args ...interface{}) *Row { rows, err := s.QueryContext(ctx, args...) if err != nil { return &Row{err: err} } return &Row{rows: rows} } // QueryRow executes a prepared query statement with the given arguments. // If an error occurs during the execution of the statement, that error will // be returned by a call to Scan on the returned *Row, which is always non-nil. // If the query selects no rows, the *Row's Scan will return ErrNoRows. // Otherwise, the *Row's Scan scans the first selected row and discards // the rest. // // Example usage: // // var name string // err := nameByUseridStmt.QueryRow(id).Scan(&name) func (s *Stmt) QueryRow(args ...interface{}) *Row { return s.QueryRowContext(context.Background(), args...) } // Close closes the statement. func (s *Stmt) Close() error { s.closemu.Lock() defer s.closemu.Unlock() if s.stickyErr != nil { return s.stickyErr } s.mu.Lock() if s.closed { s.mu.Unlock() return nil } s.closed = true s.mu.Unlock() if s.tx != nil { return s.txds.Close() } return s.db.removeDep(s, s) } func (s *Stmt) finalClose() error { s.mu.Lock() defer s.mu.Unlock() if s.css != nil { for _, v := range s.css { s.db.noteUnusedDriverStatement(v.dc, v.ds) v.dc.removeOpenStmt(v.ds) } s.css = nil } return nil } // Rows is the result of a query. Its cursor starts before the first row // of the result set. Use Next to advance through the rows: // // rows, err := db.Query("SELECT ...") // ... // defer rows.Close() // for rows.Next() { // var id int // var name string // err = rows.Scan(&id, &name) // ... // } // err = rows.Err() // get any error encountered during iteration // ... type Rows struct { dc *driverConn // owned; must call releaseConn when closed to release releaseConn func(error) rowsi driver.Rows cancel func() // called when Rows is closed, may be nil. closeStmt *driverStmt // if non-nil, statement to Close on close // closemu prevents Rows from closing while there // is an active streaming result. It is held for read during non-close operations // and exclusively during close. // // closemu guards lasterr and closed. closemu sync.RWMutex closed bool lasterr error // non-nil only if closed is true // lastcols is only used in Scan, Next, and NextResultSet which are expected // not not be called concurrently. lastcols []driver.Value } func (rs *Rows) initContextClose(ctx context.Context) { ctx, rs.cancel = context.WithCancel(ctx) go rs.awaitDone(ctx) } // awaitDone blocks until the rows are closed or the context canceled. func (rs *Rows) awaitDone(ctx context.Context) { <-ctx.Done() rs.close(ctx.Err()) } // Next prepares the next result row for reading with the Scan method. It // returns true on success, or false if there is no next result row or an error // happened while preparing it. Err should be consulted to distinguish between // the two cases. // // Every call to Scan, even the first one, must be preceded by a call to Next. func (rs *Rows) Next() bool { var doClose, ok bool withLock(rs.closemu.RLocker(), func() { doClose, ok = rs.nextLocked() }) if doClose { rs.Close() } return ok } func (rs *Rows) nextLocked() (doClose, ok bool) { if rs.closed { return false, false } if rs.lastcols == nil { rs.lastcols = make([]driver.Value, len(rs.rowsi.Columns())) } rs.lasterr = rs.rowsi.Next(rs.lastcols) if rs.lasterr != nil { // Close the connection if there is a driver error. if rs.lasterr != io.EOF { return true, false } nextResultSet, ok := rs.rowsi.(driver.RowsNextResultSet) if !ok { return true, false } // The driver is at the end of the current result set. // Test to see if there is another result set after the current one. // Only close Rows if there is no further result sets to read. if !nextResultSet.HasNextResultSet() { doClose = true } return doClose, false } return false, true } // NextResultSet prepares the next result set for reading. It returns true if // there is further result sets, or false if there is no further result set // or if there is an error advancing to it. The Err method should be consulted // to distinguish between the two cases. // // After calling NextResultSet, the Next method should always be called before // scanning. If there are further result sets they may not have rows in the result // set. func (rs *Rows) NextResultSet() bool { var doClose bool defer func() { if doClose { rs.Close() } }() rs.closemu.RLock() defer rs.closemu.RUnlock() if rs.closed { return false } rs.lastcols = nil nextResultSet, ok := rs.rowsi.(driver.RowsNextResultSet) if !ok { doClose = true return false } rs.lasterr = nextResultSet.NextResultSet() if rs.lasterr != nil { doClose = true return false } return true } // Err returns the error, if any, that was encountered during iteration. // Err may be called after an explicit or implicit Close. func (rs *Rows) Err() error { rs.closemu.RLock() defer rs.closemu.RUnlock() if rs.lasterr == io.EOF { return nil } return rs.lasterr } // Columns returns the column names. // Columns returns an error if the rows are closed, or if the rows // are from QueryRow and there was a deferred error. func (rs *Rows) Columns() ([]string, error) { rs.closemu.RLock() defer rs.closemu.RUnlock() if rs.closed { return nil, errors.New("sql: Rows are closed") } if rs.rowsi == nil { return nil, errors.New("sql: no Rows available") } return rs.rowsi.Columns(), nil } // ColumnTypes returns column information such as column type, length, // and nullable. Some information may not be available from some drivers. func (rs *Rows) ColumnTypes() ([]*ColumnType, error) { rs.closemu.RLock() defer rs.closemu.RUnlock() if rs.closed { return nil, errors.New("sql: Rows are closed") } if rs.rowsi == nil { return nil, errors.New("sql: no Rows available") } return rowsColumnInfoSetup(rs.rowsi), nil } // ColumnType contains the name and type of a column. type ColumnType struct { name string hasNullable bool hasLength bool hasPrecisionScale bool nullable bool length int64 databaseType string precision int64 scale int64 scanType reflect.Type } // Name returns the name or alias of the column. func (ci *ColumnType) Name() string { return ci.name } // Length returns the column type length for variable length column types such // as text and binary field types. If the type length is unbounded the value will // be math.MaxInt64 (any database limits will still apply). // If the column type is not variable length, such as an int, or if not supported // by the driver ok is false. func (ci *ColumnType) Length() (length int64, ok bool) { return ci.length, ci.hasLength } // DecimalSize returns the scale and precision of a decimal type. // If not applicable or if not supported ok is false. func (ci *ColumnType) DecimalSize() (precision, scale int64, ok bool) { return ci.precision, ci.scale, ci.hasPrecisionScale } // ScanType returns a Go type suitable for scanning into using Rows.Scan. // If a driver does not support this property ScanType will return // the type of an empty interface. func (ci *ColumnType) ScanType() reflect.Type { return ci.scanType } // Nullable returns whether the column may be null. // If a driver does not support this property ok will be false. func (ci *ColumnType) Nullable() (nullable, ok bool) { return ci.nullable, ci.hasNullable } // DatabaseTypeName returns the database system name of the column type. If an empty // string is returned the driver type name is not supported. // Consult your driver documentation for a list of driver data types. Length specifiers // are not included. // Common type include "VARCHAR", "TEXT", "NVARCHAR", "DECIMAL", "BOOL", "INT", "BIGINT". func (ci *ColumnType) DatabaseTypeName() string { return ci.databaseType } func rowsColumnInfoSetup(rowsi driver.Rows) []*ColumnType { names := rowsi.Columns() list := make([]*ColumnType, len(names)) for i := range list { ci := &ColumnType{ name: names[i], } list[i] = ci if prop, ok := rowsi.(driver.RowsColumnTypeScanType); ok { ci.scanType = prop.ColumnTypeScanType(i) } else { ci.scanType = reflect.TypeOf(new(interface{})).Elem() } if prop, ok := rowsi.(driver.RowsColumnTypeDatabaseTypeName); ok { ci.databaseType = prop.ColumnTypeDatabaseTypeName(i) } if prop, ok := rowsi.(driver.RowsColumnTypeLength); ok { ci.length, ci.hasLength = prop.ColumnTypeLength(i) } if prop, ok := rowsi.(driver.RowsColumnTypeNullable); ok { ci.nullable, ci.hasNullable = prop.ColumnTypeNullable(i) } if prop, ok := rowsi.(driver.RowsColumnTypePrecisionScale); ok { ci.precision, ci.scale, ci.hasPrecisionScale = prop.ColumnTypePrecisionScale(i) } } return list } // Scan copies the columns in the current row into the values pointed // at by dest. The number of values in dest must be the same as the // number of columns in Rows. // // Scan converts columns read from the database into the following // common Go types and special types provided by the sql package: // // *string // *[]byte // *int, *int8, *int16, *int32, *int64 // *uint, *uint8, *uint16, *uint32, *uint64 // *bool // *float32, *float64 // *interface{} // *RawBytes // any type implementing Scanner (see Scanner docs) // // In the most simple case, if the type of the value from the source // column is an integer, bool or string type T and dest is of type *T, // Scan simply assigns the value through the pointer. // // Scan also converts between string and numeric types, as long as no // information would be lost. While Scan stringifies all numbers // scanned from numeric database columns into *string, scans into // numeric types are checked for overflow. For example, a float64 with // value 300 or a string with value "300" can scan into a uint16, but // not into a uint8, though float64(255) or "255" can scan into a // uint8. One exception is that scans of some float64 numbers to // strings may lose information when stringifying. In general, scan // floating point columns into *float64. // // If a dest argument has type *[]byte, Scan saves in that argument a // copy of the corresponding data. The copy is owned by the caller and // can be modified and held indefinitely. The copy can be avoided by // using an argument of type *RawBytes instead; see the documentation // for RawBytes for restrictions on its use. // // If an argument has type *interface{}, Scan copies the value // provided by the underlying driver without conversion. When scanning // from a source value of type []byte to *interface{}, a copy of the // slice is made and the caller owns the result. // // Source values of type time.Time may be scanned into values of type // *time.Time, *interface{}, *string, or *[]byte. When converting to // the latter two, time.Format3339Nano is used. // // Source values of type bool may be scanned into types *bool, // *interface{}, *string, *[]byte, or *RawBytes. // // For scanning into *bool, the source may be true, false, 1, 0, or // string inputs parseable by strconv.ParseBool. func (rs *Rows) Scan(dest ...interface{}) error { rs.closemu.RLock() if rs.closed { rs.closemu.RUnlock() return errors.New("sql: Rows are closed") } rs.closemu.RUnlock() if rs.lastcols == nil { return errors.New("sql: Scan called without calling Next") } if len(dest) != len(rs.lastcols) { return fmt.Errorf("sql: expected %d destination arguments in Scan, not %d", len(rs.lastcols), len(dest)) } for i, sv := range rs.lastcols { err := convertAssign(dest[i], sv) if err != nil { return fmt.Errorf("sql: Scan error on column index %d: %v", i, err) } } return nil } // rowsCloseHook returns a function so tests may install the // hook throug a test only mutex. var rowsCloseHook = func() func(*Rows, *error) { return nil } // Close closes the Rows, preventing further enumeration. If Next is called // and returns false and there are no further result sets, // the Rows are closed automatically and it will suffice to check the // result of Err. Close is idempotent and does not affect the result of Err. func (rs *Rows) Close() error { return rs.close(nil) } func (rs *Rows) close(err error) error { rs.closemu.Lock() defer rs.closemu.Unlock() if rs.closed { return nil } rs.closed = true if rs.lasterr == nil { rs.lasterr = err } err = rs.rowsi.Close() if fn := rowsCloseHook(); fn != nil { fn(rs, &err) } if rs.cancel != nil { rs.cancel() } if rs.closeStmt != nil { rs.closeStmt.Close() } rs.releaseConn(err) return err } // Row is the result of calling QueryRow to select a single row. type Row struct { // One of these two will be non-nil: err error // deferred error for easy chaining rows *Rows } // Scan copies the columns from the matched row into the values // pointed at by dest. See the documentation on Rows.Scan for details. // If more than one row matches the query, // Scan uses the first row and discards the rest. If no row matches // the query, Scan returns ErrNoRows. func (r *Row) Scan(dest ...interface{}) error { if r.err != nil { return r.err } // TODO(bradfitz): for now we need to defensively clone all // []byte that the driver returned (not permitting // *RawBytes in Rows.Scan), since we're about to close // the Rows in our defer, when we return from this function. // the contract with the driver.Next(...) interface is that it // can return slices into read-only temporary memory that's // only valid until the next Scan/Close. But the TODO is that // for a lot of drivers, this copy will be unnecessary. We // should provide an optional interface for drivers to // implement to say, "don't worry, the []bytes that I return // from Next will not be modified again." (for instance, if // they were obtained from the network anyway) But for now we // don't care. defer r.rows.Close() for _, dp := range dest { if _, ok := dp.(*RawBytes); ok { return errors.New("sql: RawBytes isn't allowed on Row.Scan") } } if !r.rows.Next() { if err := r.rows.Err(); err != nil { return err } return ErrNoRows } err := r.rows.Scan(dest...) if err != nil { return err } // Make sure the query can be processed to completion with no errors. if err := r.rows.Close(); err != nil { return err } return nil } // A Result summarizes an executed SQL command. type Result interface { // LastInsertId returns the integer generated by the database // in response to a command. Typically this will be from an // "auto increment" column when inserting a new row. Not all // databases support this feature, and the syntax of such // statements varies. LastInsertId() (int64, error) // RowsAffected returns the number of rows affected by an // update, insert, or delete. Not every database or database // driver may support this. RowsAffected() (int64, error) } type driverResult struct { sync.Locker // the *driverConn resi driver.Result } func (dr driverResult) LastInsertId() (int64, error) { dr.Lock() defer dr.Unlock() return dr.resi.LastInsertId() } func (dr driverResult) RowsAffected() (int64, error) { dr.Lock() defer dr.Unlock() return dr.resi.RowsAffected() } func stack() string { var buf [2 << 10]byte return string(buf[:runtime.Stack(buf[:], false)]) } // withLock runs while holding lk. func withLock(lk sync.Locker, fn func()) { lk.Lock() defer lk.Unlock() // in case fn panics fn() }