Run tests parallel (#1221)

Speeds up test runner from ~10.5 seconds to ~2.8 seconds, which is not a
bad win.

Author: arp242

array_test.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"database/sql"
 	"database/sql/driver"
+	"fmt"
 	"math/rand"
 	"reflect"
 	"strings"
@@ -1654,3 +1655,57 @@ func TestArrayValueBackend(t *testing.T) {
 		}
 	}
 }
+
+func TestArrayArg(t *testing.T) {
+	db := pqtest.MustDB(t)
+	defer db.Close()
+
+	tests := []struct {
+		pgType  string
+		in, out any
+	}{
+		{"int[]", []int{245, 231}, []int64{245, 231}},
+		{"int[]", &[]int{245, 231}, []int64{245, 231}},
+		{"int[]", []int64{245, 231}, nil},
+		{"int[]", &[]int64{245, 231}, []int64{245, 231}},
+		{"varchar[]", []string{"hello", "world"}, nil},
+		{"varchar[]", &[]string{"hello", "world"}, []string{"hello", "world"}},
+	}
+
+	for _, tt := range tests {
+		t.Run("", func(t *testing.T) {
+			if tt.out == nil {
+				tt.out = tt.in
+			}
+
+			r, err := db.Query(fmt.Sprintf("SELECT $1::%s", tt.pgType), tt.in)
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer r.Close()
+
+			if !r.Next() {
+				if r.Err() != nil {
+					t.Fatal(r.Err())
+				}
+				t.Fatal("expected row")
+			}
+
+			defer func() {
+				if r.Next() {
+					t.Fatal("unexpected row")
+				}
+			}()
+
+			have := reflect.New(reflect.TypeOf(tt.out))
+			if err := r.Scan(Array(have.Interface())); err != nil {
+				t.Fatal(err)
+			}
+
+			if !reflect.DeepEqual(tt.out, have.Elem().Interface()) {
+				t.Errorf("\nhave: %v\nwant %v", have, tt.out)
+			}
+		})
+	}
+
+}

auth/kerberos/krb.go

@@ -5,12 +5,9 @@ import (
 	"strings"
 )
 
-/*
- *  Find the A record associated with a hostname
- *  In general, hostnames supplied to the driver should be
- *  canonicalized because the KDC usually only has one
- *  principal and not one per potential alias of a host.
- */
+// Find the A record associated with a hostname In general, hostnames supplied
+// to the driver should be canonicalized because the KDC usually only has one
+// principal and not one per potential alias of a host.
 func canonicalizeHostname(host string) (string, error) {
 	canon := host
 

auth/kerberos/krb_unix.go

@@ -1,5 +1,7 @@
 //go:build !windows
 
+// UNIX Kerberos support, using jcmturner's pure-go implementation
+
 package kerberos
 
 import (
@@ -14,11 +16,6 @@ import (
 	"github.com/jcmturner/gokrb5/v8/spnego"
 )
 
-/*
- * UNIX Kerberos support, using jcmturner's pure-go
- * implementation
- */
-
 // GSS implements the pq.GSS interface.
 type GSS struct {
 	cli *client.Client

cmd/pqlisten/main.go

@@ -0,0 +1,93 @@
+// Command pqlisten is a self-contained Go program which uses the LISTEN /
+// NOTIFY mechanism to avoid polling the database while waiting for more work to
+// arrive.
+//
+// You can see the program in action by defining a function similar to the
+// following:
+//
+//	CREATE OR REPLACE FUNCTION public.get_work()
+//	  RETURNS bigint
+//	  LANGUAGE sql
+//	  AS $$
+//	    SELECT CASE WHEN random() >= 0.2 THEN int8 '1' END
+//	  $$
+//	;
+package main
+
+import (
+	"database/sql"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/lib/pq"
+)
+
+func main() {
+	db, err := sql.Open("postgres", strings.Join(os.Args[1:], " "))
+	if err != nil {
+		panic(err)
+	}
+
+	reportProblem := func(ev pq.ListenerEventType, err error) {
+		if err != nil {
+			fmt.Fprintln(os.Stderr, err.Error())
+		}
+	}
+
+	var (
+		minReconn = 10 * time.Second
+		maxReconn = time.Minute
+		listener  = pq.NewListener("", minReconn, maxReconn, reportProblem)
+	)
+	err = listener.Listen("getwork")
+	if err != nil {
+		panic(err)
+	}
+
+	fmt.Println("entering main loop")
+	for {
+		// process all available work before waiting for notifications
+		getWork(db)
+		waitForNotification(listener)
+	}
+}
+
+func doWork(db *sql.DB, work int64) {
+	// work here
+	fmt.Println("doing work...")
+}
+
+func getWork(db *sql.DB) {
+	for {
+		// get work from the database here
+		var work sql.NullInt64
+		err := db.QueryRow("SELECT get_work()").Scan(&work)
+		if err != nil {
+			fmt.Println("call to get_work() failed: ", err)
+			time.Sleep(10 * time.Second)
+			continue
+		}
+		if !work.Valid {
+			// no more work to do
+			fmt.Println("ran out of work")
+			return
+		}
+
+		fmt.Println("starting work on ", work.Int64)
+		go doWork(db, work.Int64)
+	}
+}
+
+func waitForNotification(l *pq.Listener) {
+	select {
+	case <-l.Notify:
+		fmt.Println("received notification, new work available")
+	case <-time.After(90 * time.Second):
+		go l.Ping()
+		// Check if there's more work available, just in case it takes a while
+		// for the Listener to notice connection loss and reconnect.
+		fmt.Println("received no work for 90 seconds, checking for new work")
+	}
+}

conn.go

@@ -17,6 +17,7 @@ import (
 	"os"
 	"os/user"
 	"path/filepath"
+	"reflect"
 	"runtime"
 	"strconv"
 	"strings"
@@ -45,8 +46,13 @@ var (
 
 // Compile time validation that our types implement the expected interfaces
 var (
-	_ driver.Driver    = Driver{}
-	_ driver.Validator = &conn{}
+	_ driver.Driver            = Driver{}
+	_ driver.Validator         = (*conn)(nil)
+	_ driver.NamedValueChecker = (*conn)(nil)
+	_ driver.Pinger            = (*conn)(nil)
+	_ driver.SessionResetter   = (*conn)(nil)
+	_ driver.ExecerContext     = (*conn)(nil)
+	_ driver.QueryerContext    = (*conn)(nil)
 )
 
 // Driver is the Postgres database driver.
@@ -924,6 +930,31 @@ func toNamedValue(v []driver.Value) []driver.NamedValue {
 	return v2
 }
 
+// CheckNamedValue implements [driver.NamedValueChecker].
+func (c *conn) CheckNamedValue(nv *driver.NamedValue) error {
+	if _, ok := nv.Value.(driver.Valuer); ok {
+		// Ignore Valuer, for backward compatibility with pq.Array().
+		return driver.ErrSkip
+	}
+
+	// Ignoring []byte / []uint8.
+	if _, ok := nv.Value.([]uint8); ok {
+		return driver.ErrSkip
+	}
+
+	v := reflect.ValueOf(nv.Value)
+	if v.Kind() == reflect.Ptr {
+		v = v.Elem()
+	}
+	if v.Kind() == reflect.Slice {
+		var err error
+		nv.Value, err = Array(v.Interface()).Value()
+		return err
+	}
+
+	return driver.ErrSkip
+}
+
 // Implement the "Queryer" interface
 func (cn *conn) Query(query string, args []driver.Value) (driver.Rows, error) {
 	return cn.query(query, toNamedValue(args))
@@ -2141,11 +2172,6 @@ func alnumLowerASCII(ch rune) rune {
 	return -1 // discard
 }
 
-// The database/sql/driver package says:
-// All Conn implementations should implement the following interfaces: Pinger, SessionResetter, and Validator.
-var _ driver.Pinger = &conn{}
-var _ driver.SessionResetter = &conn{}
-
 func (cn *conn) ResetSession(ctx context.Context) error {
 	// Ensure bad connections are reported: From database/sql/driver:
 	// If a connection is never returned to the connection pool but immediately reused, then

conn_go18.go

@@ -9,9 +9,7 @@ import (
 	"time"
 )
 
-const (
-	watchCancelDialContextTimeout = time.Second * 10
-)
+const watchCancelDialContextTimeout = 10 * time.Second
 
 // Implement the "QueryerContext" interface
 func (cn *conn) QueryContext(ctx context.Context, query string, args []driver.NamedValue) (driver.Rows, error) {