New DBAction for creating index concurrently (#939)

Relates to #742

Author: kvch

pkg/migrations/dbactions.go

@@ -220,6 +220,80 @@ func (a *dropFunctionAction) Execute(ctx context.Context) error {
 	return err
 }
 
+type createIndexConcurrentlyAction struct {
+	conn              db.DB
+	table             string
+	name              string
+	method            string
+	unique            bool
+	columns           map[string]IndexField
+	storageParameters string
+	predicate         string
+}
+
+func NewCreateIndexConcurrentlyAction(conn db.DB, table, name, method string, unique bool, columns map[string]IndexField, storageParameters, predicate string) *createIndexConcurrentlyAction {
+	return &createIndexConcurrentlyAction{
+		conn:              conn,
+		table:             table,
+		name:              name,
+		method:            method,
+		unique:            unique,
+		columns:           columns,
+		storageParameters: storageParameters,
+		predicate:         predicate,
+	}
+}
+
+func (a *createIndexConcurrentlyAction) Execute(ctx context.Context) error {
+	stmtFmt := "CREATE INDEX CONCURRENTLY %s ON %s"
+	if a.unique {
+		stmtFmt = "CREATE UNIQUE INDEX CONCURRENTLY %s ON %s"
+	}
+	stmt := fmt.Sprintf(stmtFmt,
+		pq.QuoteIdentifier(a.name),
+		pq.QuoteIdentifier(a.table))
+
+	if a.method != "" {
+		stmt += fmt.Sprintf(" USING %s", a.method)
+	}
+
+	colSQLs := make([]string, 0, len(a.columns))
+	for columnName, settings := range a.columns {
+		colSQL := pq.QuoteIdentifier(columnName)
+		// deparse collations
+		if settings.Collate != "" {
+			colSQL += " COLLATE " + settings.Collate
+		}
+		// deparse operator classes and their parameters
+		if settings.Opclass != nil {
+			colSQL += " " + settings.Opclass.Name
+			if len(settings.Opclass.Params) > 0 {
+				colSQL += " " + strings.Join(settings.Opclass.Params, ", ")
+			}
+		}
+		// deparse sort order of the index column
+		if settings.Sort != "" {
+			colSQL += " " + string(settings.Sort)
+		}
+		// deparse nulls order of the index column
+		if settings.Nulls != nil {
+			colSQL += " " + string(*settings.Nulls)
+		}
+		colSQLs = append(colSQLs, colSQL)
+	}
+	stmt += fmt.Sprintf(" (%s)", strings.Join(colSQLs, ", "))
+
+	if a.storageParameters != "" {
+		stmt += fmt.Sprintf(" WITH (%s)", a.storageParameters)
+	}
+
+	if a.predicate != "" {
+		stmt += fmt.Sprintf(" WHERE %s", a.predicate)
+	}
+	_, err := a.conn.ExecContext(ctx, stmt)
+	return err
+}
+
 // commentColumnAction is a DBAction that adds a comment to a column in a table.
 type commentColumnAction struct {
 	conn    db.DB

pkg/migrations/op_create_index.go

@@ -5,7 +5,6 @@ package migrations
 import (
 	"context"
 	"fmt"
-	"strings"
 
 	"github.com/lib/pq"
 
@@ -27,55 +26,23 @@ func (o *OpCreateIndex) Start(ctx context.Context, l Logger, conn db.DB, latestS
 		return nil, TableDoesNotExistError{Name: o.Table}
 	}
 
-	// create index concurrently
-	stmtFmt := "CREATE INDEX CONCURRENTLY %s ON %s"
-	if o.Unique {
-		stmtFmt = "CREATE UNIQUE INDEX CONCURRENTLY %s ON %s"
+	cols := make(map[string]IndexField, len(o.Columns))
+	for name, settings := range map[string]IndexField(o.Columns) {
+		physicalName := table.PhysicalColumnNamesFor(name)
+		cols[physicalName[0]] = settings
 	}
-	stmt := fmt.Sprintf(stmtFmt,
-		pq.QuoteIdentifier(o.Name),
-		pq.QuoteIdentifier(table.Name))
 
-	if o.Method != "" {
-		stmt += fmt.Sprintf(" USING %s", string(o.Method))
-	}
-
-	colSQLs := make([]string, 0, len(o.Columns))
-	for columnName, settings := range map[string]IndexField(o.Columns) {
-		physicalName := table.PhysicalColumnNamesFor(columnName)
-		colSQL := pq.QuoteIdentifier(physicalName[0])
-		// deparse collations
-		if settings.Collate != "" {
-			colSQL += " COLLATE " + settings.Collate
-		}
-		// deparse operator classes and their parameters
-		if settings.Opclass != nil {
-			colSQL += " " + settings.Opclass.Name
-			if len(settings.Opclass.Params) > 0 {
-				colSQL += " " + strings.Join(settings.Opclass.Params, ", ")
-			}
-		}
-		// deparse sort order of the index column
-		if settings.Sort != "" {
-			colSQL += " " + string(settings.Sort)
-		}
-		// deparse nulls order of the index column
-		if settings.Nulls != nil {
-			colSQL += " " + string(*settings.Nulls)
-		}
-		colSQLs = append(colSQLs, colSQL)
-	}
-	stmt += fmt.Sprintf(" (%s)", strings.Join(colSQLs, ", "))
-
-	if o.StorageParameters != "" {
-		stmt += fmt.Sprintf(" WITH (%s)", o.StorageParameters)
-	}
-
-	if o.Predicate != "" {
-		stmt += fmt.Sprintf(" WHERE %s", o.Predicate)
-	}
+	err := NewCreateIndexConcurrentlyAction(
+		conn,
+		table.Name,
+		o.Name,
+		string(o.Method),
+		o.Unique,
+		cols,
+		o.StorageParameters,
+		o.Predicate,
+	).Execute(ctx)
 
-	_, err := conn.ExecContext(ctx, stmt)
 	return nil, err
 }