Return list of DBAction from Start (#944)

This PR returns adds a new return value to `Start` to return the list of
`DBActions`. However, returning 3 values is too many, so in a follow-up
PR I am combining the list of `DBActions` and `backfill.Task` into the
same data structure. The name is pending, my ideas so far `StartStep` or
`StartOperation`. If you have a better name, I am open to using them.

New data structure in a follow-up PR:
```golang
type StartOperation struct {
    Actions []DBAction
    Task *backfill.Task
}
```

---------

Co-authored-by: Andrew Farries <andyrb@gmail.com>

Author: kvch

pkg/migrations/migrations.go

@@ -16,11 +16,11 @@ import (
 
 // Operation is an operation that can be applied to a schema
 type Operation interface {
-	// Start will apply the required changes to enable supporting the new schema
+	// Start will return the list of required changes to enable supporting the new schema
 	// version in the database (through a view)
 	// update the given views to expose the new schema version
 	// Returns the table that requires backfilling, if any.
-	Start(ctx context.Context, l Logger, conn db.DB, s *schema.Schema) (*backfill.Task, error)
+	Start(ctx context.Context, l Logger, conn db.DB, s *schema.Schema) ([]DBAction, *backfill.Task, error)
 
 	// Complete will update the database schema to match the current version
 	// after calling Start.
@@ -118,7 +118,7 @@ func (m *Migration) UpdateVirtualSchema(ctx context.Context, s *schema.Schema) e
 	// Run `Start` on each operation using the fake DB. Updates will be made to
 	// the in-memory schema `s` without touching the physical database.
 	for _, op := range m.Operations {
-		if _, err := op.Start(ctx, NewNoopLogger(), db, s); err != nil {
+		if _, _, err := op.Start(ctx, NewNoopLogger(), db, s); err != nil {
 			return err
 		}
 	}

pkg/migrations/op_add_column.go

@@ -19,12 +19,12 @@ var (
 	_ Createable = (*OpAddColumn)(nil)
 )
 
-func (o *OpAddColumn) Start(ctx context.Context, l Logger, conn db.DB, s *schema.Schema) (*backfill.Task, error) {
+func (o *OpAddColumn) Start(ctx context.Context, l Logger, conn db.DB, s *schema.Schema) ([]DBAction, *backfill.Task, error) {
 	l.LogOperationStart(o)
 
 	table := s.GetTable(o.Table)
 	if table == nil {
-		return nil, TableDoesNotExistError{Name: o.Table}
+		return nil, nil, TableDoesNotExistError{Name: o.Table}
 	}
 
 	// If the column has a DEFAULT, check if it can be added using the fast path
@@ -33,20 +33,19 @@ func (o *OpAddColumn) Start(ctx context.Context, l Logger, conn db.DB, s *schema
 	if o.Column.HasDefault() {
 		v, err := defaults.UsesFastPath(ctx, conn, table.Name, o.Column.Type, *o.Column.Default)
 		if err != nil {
-			return nil, fmt.Errorf("failed to check for fast path default optimization: %w", err)
+			return nil, nil, fmt.Errorf("failed to check for fast path default optimization: %w", err)
 		}
 		fastPathDefault = v
 	}
 
-	if err := addColumn(ctx, conn, *o, table, fastPathDefault); err != nil {
-		return nil, fmt.Errorf("failed to start add column operation: %w", err)
+	action, err := addColumn(conn, *o, table, fastPathDefault)
+	if err != nil {
+		return nil, nil, err
 	}
+	dbActions := []DBAction{action}
 
 	if o.Column.Comment != nil {
-		err := NewCommentColumnAction(conn, table.Name, TemporaryName(o.Column.Name), o.Column.Comment).Execute(ctx)
-		if err != nil {
-			return nil, fmt.Errorf("failed to add comment to column: %w", err)
-		}
+		dbActions = append(dbActions, NewCommentColumnAction(conn, table.Name, TemporaryName(o.Column.Name), o.Column.Comment))
 	}
 
 	// If the column is `NOT NULL` and there is no default value (either because
@@ -56,47 +55,48 @@ func (o *OpAddColumn) Start(ctx context.Context, l Logger, conn db.DB, s *schema
 	skipInherit := false
 	skipValidate := true
 	if !o.Column.IsNullable() && (o.Column.Default == nil || !fastPathDefault) {
-		if err := NewCreateCheckConstraintAction(
-			conn,
-			table.Name,
-			NotNullConstraintName(o.Column.Name),
-			fmt.Sprintf("%s IS NOT NULL", o.Column.Name),
-			[]string{o.Column.Name},
-			skipInherit,
-			skipValidate,
-		).Execute(ctx); err != nil {
-			return nil, fmt.Errorf("failed to add not null constraint: %w", err)
-		}
+		dbActions = append(dbActions,
+			NewCreateCheckConstraintAction(
+				conn,
+				table.Name,
+				NotNullConstraintName(o.Column.Name),
+				fmt.Sprintf("%s IS NOT NULL", o.Column.Name),
+				[]string{o.Column.Name},
+				skipInherit,
+				skipValidate,
+			))
 	}
 
 	if o.Column.Check != nil {
-		if err := NewCreateCheckConstraintAction(
-			conn,
-			table.Name,
-			o.Column.Check.Name,
-			o.Column.Check.Constraint,
-			[]string{o.Column.Name},
-			skipInherit,
-			skipValidate,
-		).Execute(ctx); err != nil {
-			return nil, fmt.Errorf("failed to add check constraint: %w", err)
-		}
+		dbActions = append(dbActions,
+			NewCreateCheckConstraintAction(
+				conn,
+				table.Name,
+				o.Column.Check.Name,
+				o.Column.Check.Constraint,
+				[]string{o.Column.Name},
+				skipInherit,
+				skipValidate,
+			))
 	}
 
 	if o.Column.Unique {
-		createIndex := NewCreateUniqueIndexConcurrentlyAction(conn, s.Name, UniqueIndexName(o.Column.Name), table.Name, TemporaryName(o.Column.Name))
-		err := createIndex.Execute(ctx)
-		if err != nil {
-			return nil, fmt.Errorf("failed to add unique index: %w", err)
-		}
+		dbActions = append(dbActions,
+			NewCreateUniqueIndexConcurrentlyAction(
+				conn,
+				s.Name,
+				UniqueIndexName(o.Column.Name),
+				table.Name,
+				TemporaryName(o.Column.Name),
+			))
 	}
 
 	// If the column has a DEFAULT that cannot be set using the fast path
 	// optimization, the `up` SQL expression must be used to set the DEFAULT
 	// value for the column.
 	if o.Column.HasDefault() && !fastPathDefault {
 		if o.Up != *o.Column.Default {
-			return nil, UpSQLMustBeColumnDefaultError{Column: o.Column.Name}
+			return nil, nil, UpSQLMustBeColumnDefaultError{Column: o.Column.Name}
 		}
 	}
 
@@ -118,7 +118,7 @@ func (o *OpAddColumn) Start(ctx context.Context, l Logger, conn db.DB, s *schema
 	tmpColumn.Name = TemporaryName(o.Column.Name)
 	table.AddColumn(o.Column.Name, tmpColumn)
 
-	return task, nil
+	return dbActions, task, nil
 }
 
 func toSchemaColumn(c Column) *schema.Column {
@@ -253,7 +253,7 @@ func (o *OpAddColumn) Validate(ctx context.Context, s *schema.Schema) error {
 	return nil
 }
 
-func addColumn(ctx context.Context, conn db.DB, o OpAddColumn, t *schema.Table, fastPathDefault bool) error {
+func addColumn(conn db.DB, o OpAddColumn, t *schema.Table, fastPathDefault bool) (DBAction, error) {
 	// don't add non-nullable columns with no default directly
 	// they are handled by:
 	// - adding the column as nullable
@@ -266,7 +266,7 @@ func addColumn(ctx context.Context, conn db.DB, o OpAddColumn, t *schema.Table,
 	}
 
 	if o.Column.Generated != nil {
-		return fmt.Errorf("adding generated columns to existing tables is not supported")
+		return nil, fmt.Errorf("adding generated columns to existing tables is not supported")
 	}
 
 	// Don't add a column with a CHECK constraint directly.
@@ -299,7 +299,7 @@ func addColumn(ctx context.Context, conn db.DB, o OpAddColumn, t *schema.Table,
 	o.Column.Name = TemporaryName(o.Column.Name)
 
 	withPK := true
-	return NewAddColumnAction(conn, t.Name, o.Column, withPK).Execute(ctx)
+	return NewAddColumnAction(conn, t.Name, o.Column, withPK), nil
 }
 
 // upgradeNotNullConstraintToNotNullAttribute validates and upgrades a NOT NULL

pkg/migrations/op_alter_column.go

@@ -18,24 +18,24 @@ var (
 	_ Createable = (*OpAlterColumn)(nil)
 )
 
-func (o *OpAlterColumn) Start(ctx context.Context, l Logger, conn db.DB, s *schema.Schema) (*backfill.Task, error) {
+func (o *OpAlterColumn) Start(ctx context.Context, l Logger, conn db.DB, s *schema.Schema) ([]DBAction, *backfill.Task, error) {
 	l.LogOperationStart(o)
 
 	table := s.GetTable(o.Table)
 	if table == nil {
-		return nil, TableDoesNotExistError{Name: o.Table}
+		return nil, nil, TableDoesNotExistError{Name: o.Table}
 	}
 	column := table.GetColumn(o.Column)
 	if column == nil {
-		return nil, ColumnDoesNotExistError{Table: o.Table, Name: o.Column}
+		return nil, nil, ColumnDoesNotExistError{Table: o.Table, Name: o.Column}
 	}
 	ops := o.subOperations()
 
 	// Duplicate the column on the underlying table.
 	d := duplicatorForOperations(ops, conn, table, column).
 		WithName(column.Name, TemporaryName(o.Column))
 	if err := d.Execute(ctx); err != nil {
-		return nil, fmt.Errorf("failed to duplicate column: %w", err)
+		return nil, nil, fmt.Errorf("failed to duplicate column: %w", err)
 	}
 
 	// Copy the columns from table columns, so we can use it later
@@ -79,16 +79,19 @@ func (o *OpAlterColumn) Start(ctx context.Context, l Logger, conn db.DB, s *sche
 		},
 	)
 	task := backfill.NewTask(table, triggers...)
+
+	var dbActions []DBAction
 	// perform any operation specific start steps
 	for _, op := range ops {
-		bf, err := op.Start(ctx, l, conn, s)
+		actions, bf, err := op.Start(ctx, l, conn, s)
 		if err != nil {
-			return nil, err
+			return nil, nil, err
 		}
 		task.AddTriggers(bf)
+		dbActions = append(dbActions, actions...)
 	}
 
-	return task, nil
+	return dbActions, task, nil
 }
 
 func (o *OpAlterColumn) Complete(l Logger, conn db.DB, s *schema.Schema) ([]DBAction, error) {

pkg/migrations/op_change_type.go

@@ -20,15 +20,15 @@ type OpChangeType struct {
 
 var _ Operation = (*OpChangeType)(nil)
 
-func (o *OpChangeType) Start(ctx context.Context, l Logger, conn db.DB, s *schema.Schema) (*backfill.Task, error) {
+func (o *OpChangeType) Start(ctx context.Context, l Logger, conn db.DB, s *schema.Schema) ([]DBAction, *backfill.Task, error) {
 	l.LogOperationStart(o)
 
 	table := s.GetTable(o.Table)
 	if table == nil {
-		return nil, TableDoesNotExistError{Name: o.Table}
+		return nil, nil, TableDoesNotExistError{Name: o.Table}
 	}
 
-	return backfill.NewTask(table), nil
+	return nil, backfill.NewTask(table), nil
 }
 
 func (o *OpChangeType) Complete(l Logger, conn db.DB, s *schema.Schema) ([]DBAction, error) {

pkg/migrations/op_create_constraint.go

@@ -4,7 +4,6 @@ package migrations
 
 import (
 	"context"
-	"fmt"
 
 	"github.com/xataio/pgroll/pkg/backfill"
 	"github.com/xataio/pgroll/pkg/db"
@@ -16,19 +15,19 @@ var (
 	_ Createable = (*OpCreateConstraint)(nil)
 )
 
-func (o *OpCreateConstraint) Start(ctx context.Context, l Logger, conn db.DB, s *schema.Schema) (*backfill.Task, error) {
+func (o *OpCreateConstraint) Start(ctx context.Context, l Logger, conn db.DB, s *schema.Schema) ([]DBAction, *backfill.Task, error) {
 	l.LogOperationStart(o)
 
 	table := s.GetTable(o.Table)
 	if table == nil {
-		return nil, TableDoesNotExistError{Name: o.Table}
+		return nil, nil, TableDoesNotExistError{Name: o.Table}
 	}
 
 	columns := make([]*schema.Column, len(o.Columns))
 	for i, colName := range o.Columns {
 		columns[i] = table.GetColumn(colName)
 		if columns[i] == nil {
-			return nil, ColumnDoesNotExistError{Table: o.Table, Name: colName}
+			return nil, nil, ColumnDoesNotExistError{Table: o.Table, Name: colName}
 		}
 	}
 
@@ -37,9 +36,7 @@ func (o *OpCreateConstraint) Start(ctx context.Context, l Logger, conn db.DB, s
 	for _, colName := range o.Columns {
 		d = d.WithName(table.GetColumn(colName).Name, TemporaryName(colName))
 	}
-	if err := d.Execute(ctx); err != nil {
-		return nil, fmt.Errorf("failed to duplicate columns for new constraint: %w", err)
-	}
+	dbActions := []DBAction{d}
 
 	// Copy the columns from table columns, so we can use it later
 	// in the down trigger with the physical name
@@ -89,14 +86,25 @@ func (o *OpCreateConstraint) Start(ctx context.Context, l Logger, conn db.DB, s
 
 	switch o.Type {
 	case OpCreateConstraintTypeUnique, OpCreateConstraintTypePrimaryKey:
-		return task, NewCreateUniqueIndexConcurrentlyAction(conn, s.Name, o.Name, table.Name, temporaryNames(o.Columns)...).Execute(ctx)
+		dbActions = append(dbActions,
+			NewCreateUniqueIndexConcurrentlyAction(conn, s.Name, o.Name, table.Name, temporaryNames(o.Columns)...),
+		)
+		return dbActions, task, nil
+
 	case OpCreateConstraintTypeCheck:
-		return task, NewCreateCheckConstraintAction(conn, table.Name, o.Name, *o.Check, o.Columns, o.NoInherit, true).Execute(ctx)
+		dbActions = append(dbActions,
+			NewCreateCheckConstraintAction(conn, table.Name, o.Name, *o.Check, o.Columns, o.NoInherit, true),
+		)
+		return dbActions, task, nil
+
 	case OpCreateConstraintTypeForeignKey:
-		return task, NewCreateFKConstraintAction(conn, table.Name, o.Name, temporaryNames(o.Columns), o.References, false, false, true).Execute(ctx)
+		dbActions = append(dbActions,
+			NewCreateFKConstraintAction(conn, table.Name, o.Name, temporaryNames(o.Columns), o.References, false, false, true),
+		)
+		return dbActions, task, nil
 	}
 
-	return task, nil
+	return dbActions, task, nil
 }
 
 func (o *OpCreateConstraint) Complete(l Logger, conn db.DB, s *schema.Schema) ([]DBAction, error) {

pkg/migrations/op_create_index.go

@@ -18,12 +18,12 @@ var (
 	_ Createable = (*OpCreateIndex)(nil)
 )
 
-func (o *OpCreateIndex) Start(ctx context.Context, l Logger, conn db.DB, s *schema.Schema) (*backfill.Task, error) {
+func (o *OpCreateIndex) Start(ctx context.Context, l Logger, conn db.DB, s *schema.Schema) ([]DBAction, *backfill.Task, error) {
 	l.LogOperationStart(o)
 
 	table := s.GetTable(o.Table)
 	if table == nil {
-		return nil, TableDoesNotExistError{Name: o.Table}
+		return nil, nil, TableDoesNotExistError{Name: o.Table}
 	}
 
 	cols := make(map[string]IndexField, len(o.Columns))
@@ -32,18 +32,20 @@ func (o *OpCreateIndex) Start(ctx context.Context, l Logger, conn db.DB, s *sche
 		cols[physicalName[0]] = settings
 	}
 
-	err := NewCreateIndexConcurrentlyAction(
-		conn,
-		table.Name,
-		o.Name,
-		string(o.Method),
-		o.Unique,
-		cols,
-		o.StorageParameters,
-		o.Predicate,
-	).Execute(ctx)
-
-	return nil, err
+	dbActions := []DBAction{
+		NewCreateIndexConcurrentlyAction(
+			conn,
+			table.Name,
+			o.Name,
+			string(o.Method),
+			o.Unique,
+			cols,
+			o.StorageParameters,
+			o.Predicate,
+		),
+	}
+
+	return dbActions, nil, nil
 }
 
 func (o *OpCreateIndex) Complete(l Logger, conn db.DB, s *schema.Schema) ([]DBAction, error) {