[WIP] Fully parallelize query execution

cmd/pgverify/cmd.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"strings"
 
-	"github.com/jackc/pgx/v4"
+	"github.com/jackc/pgx/v4/pgxpool"
 	log "github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 
@@ -49,9 +49,9 @@ var rootCmd = &cobra.Command{
 	Long: `Verify data consistency between PostgreSQL syntax compatible databases.`,
 	Args: cobra.MinimumNArgs(1),
 	RunE: func(cmd *cobra.Command, args []string) error {
-		var targets []*pgx.ConnConfig
+		var targets []*pgxpool.Config
 		for _, target := range args {
-			connConfig, err := pgx.ParseConfig(target)
+			connConfig, err := pgxpool.ParseConfig(target)
 			if err != nil {
 				return fmt.Errorf("invalid target URI %s: %w", target, err)
 			}

go.mod

@@ -107,6 +107,7 @@ require (
 	github.com/jackc/pgproto3/v2 v2.2.0 // indirect
 	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
 	github.com/jackc/pgtype v1.10.0 // indirect
+	github.com/jackc/puddle v1.2.1 // indirect
 	github.com/jgautheron/goconst v1.6.0 // indirect
 	github.com/jingyugao/rowserrcheck v1.1.1 // indirect
 	github.com/jirfag/go-printf-func-name v0.0.0-20200119135958-7558a9eaa5af // indirect

go.sum

@@ -643,6 +643,7 @@ github.com/jackc/pgx/v4 v4.15.0/go.mod h1:D/zyOyXiaM1TmVWnOM18p0xdDtdakRBa0RsVGI
 github.com/jackc/puddle v0.0.0-20190413234325-e4ced69a3a2b/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jackc/puddle v0.0.0-20190608224051-11cab39313c9/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jackc/puddle v1.1.3/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
+github.com/jackc/puddle v1.2.1 h1:gI8os0wpRXFd4FiAY2dWiqRK037tjj3t7rKFeO4X5iw=
 github.com/jackc/puddle v1.2.1/go.mod h1:m4B5Dj62Y0fbyuIc15OsIqK0+JU8nkqQjsgx7dvjSWk=
 github.com/jgautheron/goconst v1.6.0 h1:gbMLWKRMkzAc6kYsQL6/TxaoBUg3Jm9LSF/Ih1ADWGA=
 github.com/jgautheron/goconst v1.6.0/go.mod h1:aAosetZ5zaeC/2EfMeRswtxUFBpe2Hr7HzkgX4fanO4=

integration_test.go

@@ -15,7 +15,7 @@ import (
 
 	"github.com/cjfinnell/pgverify"
 	"github.com/google/uuid"
-	"github.com/jackc/pgx/v4"
+	"github.com/jackc/pgx/v4/pgxpool"
 	"github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -27,17 +27,17 @@ var (
 	dbName     = "test"
 )
 
-func waitForDBReady(t *testing.T, ctx context.Context, config *pgx.ConnConfig) bool {
+func waitForDBReady(t *testing.T, ctx context.Context, config *pgxpool.Config) bool {
 	t.Helper()
 
 	connected := false
 
 	for count := 0; count < 30; count++ {
-		conn, err := pgx.ConnectConfig(ctx, config)
+		conn, err := pgxpool.ConnectConfig(ctx, config)
 		if err == nil {
 			connected = true
 
-			conn.Close(ctx)
+			conn.Close()
 
 			break
 		}
@@ -257,7 +257,7 @@ func TestVerifyData(t *testing.T) {
 	}
 
 	// Act
-	var targets []*pgx.ConnConfig
+	var targets []*pgxpool.Config
 
 	var aliases []string
 
@@ -266,13 +266,13 @@ func TestVerifyData(t *testing.T) {
 		// Create db and connect
 		_, port, err := createContainer(t, ctx, db.image, db.port, db.env, db.cmd)
 		require.NoError(t, err)
-		config, err := pgx.ParseConfig(fmt.Sprintf("postgresql://%s@127.0.0.1:%d%s", db.userPassword, port, db.db))
+		config, err := pgxpool.ParseConfig(fmt.Sprintf("postgresql://%s@127.0.0.1:%d%s", db.userPassword, port, db.db))
 		require.NoError(t, err)
 		assert.True(t, waitForDBReady(t, ctx, config))
-		conn, err := pgx.ConnectConfig(ctx, config)
+		conn, err := pgxpool.ConnectConfig(ctx, config)
 		require.NoError(t, err)
 
-		defer conn.Close(ctx)
+		defer conn.Close()
 
 		// Create and populate tables
 		for _, tableName := range tableNames {

results.go

@@ -38,16 +38,24 @@ func NewResults(targetNames []string, testModes []string) *Results {
 	}
 }
 
-// SingleResult represents the verification result from a single target, with the schema:
-// SingleResult[schema][table][mode] = test output.
-type SingleResult map[string]map[string]map[string]string
+// DatabaseResult represents the verification result from a single target database:
+// DatabaseResult[schema][table][mode] = test output.
+type DatabaseResult map[string]SchemaResult
+
+// SchemaResult represents the verification result from a single schema:
+// SchemaResult[table][mode] = test output.
+type SchemaResult map[string]TableResult
+
+// TableResult represents the verification result from a single table:
+// TableResult[mode] = test output.
+type TableResult map[string]string
 
 // AddResult adds a SingleResult from a test on a specific target to the Results object.
-func (r *Results) AddResult(targetName string, schemaTableHashes SingleResult) {
+func (r *Results) AddResult(targetName string, databaseHashes DatabaseResult) {
 	r.mutex.Lock()
 	defer r.mutex.Unlock()
 
-	for schema, tables := range schemaTableHashes {
+	for schema, tables := range databaseHashes {
 		if _, ok := r.content[schema]; !ok {
 			r.content[schema] = make(map[string]map[string]map[string][]string)
 		}