Add new subcommand to validate migration files (#856)

This PR adds a new subcommand to validate migration files called
`validate`.

```
Usage:
  pgroll validate <file> [flags]

Examples:
validate migrations/03_my_migration.yaml

Flags:
  -h, --help   help for validate
```

It is first step towards providing a way to validate migration files
before running them. At the moment it can validate a single migration
file against the latest schema in the database. This can be added to
your local workflow, or a linter to make sure that the newly added
migration file is correct.

Future improvements can be
* validating multiple migrations
* validating all unapplied migrations in a folder
* verifying `up` migrations on a subset of the table
#169
* run `jsonschema.Validate` during validate
#673

The PR contains some minor changes to the migration phase `Start`. Now
migration validation is decoupled from it. The previous implementation
of `Start` started a new migration before validating the migration it
got as a parameter. So this change does not interfere with the UX of
pgroll.

Closes #370

---------

Co-authored-by: Andrew Farries <[email protected]>

Author: kvch

cli-definition.json

@@ -239,6 +239,17 @@
       "flags": [],
       "subcommands": [],
       "args": []
+    },
+    {
+      "name": "validate",
+      "short": "Validate a migration file",
+      "use": "validate <file>",
+      "example": "validate migrations/03_my_migration.yaml",
+      "flags": [],
+      "subcommands": [],
+      "args": [
+        "file"
+      ]
     }
   ],
   "flags": [

cmd/root.go

@@ -107,6 +107,7 @@ func Prepare() *cobra.Command {
 	rootCmd.AddCommand(latestCmd())
 	rootCmd.AddCommand(convertCmd())
 	rootCmd.AddCommand(baselineCmd())
+	rootCmd.AddCommand(validateCmd)
 
 	return rootCmd
 }

cmd/validate.go

@@ -0,0 +1,39 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package cmd
+
+import (
+	"os"
+	"path/filepath"
+
+	"github.com/spf13/cobra"
+	"github.com/xataio/pgroll/pkg/migrations"
+)
+
+var validateCmd = &cobra.Command{
+	Use:       "validate <file>",
+	Short:     "Validate a migration file",
+	Example:   "validate migrations/03_my_migration.yaml",
+	Args:      cobra.ExactArgs(1),
+	ValidArgs: []string{"file"},
+	RunE: func(cmd *cobra.Command, args []string) error {
+		ctx := cmd.Context()
+		fileName := args[0]
+
+		m, err := NewRollWithInitCheck(ctx)
+		if err != nil {
+			return err
+		}
+		defer m.Close()
+
+		migration, err := migrations.ReadMigration(os.DirFS(filepath.Dir(fileName)), filepath.Base(fileName))
+		if err != nil {
+			return err
+		}
+		err = m.Validate(ctx, migration)
+		if err != nil {
+			return err
+		}
+		return nil
+	},
+}

docs/cli/validate.mdx

@@ -0,0 +1,18 @@
+---
+title: Validate
+description: Validate a pgroll migration
+---
+
+## Command
+
+```
+$ pgroll validate sql/03_add_column.yaml
+```
+
+This validates the migration defined in the `sql/03_add_column.yaml` file.
+
+The command can detect the following errors:
+
+* syntax error in pgroll migration format
+* unknown/invalid configuration options and settings in the migration file
+* reference to unknown database objects

docs/config.json

@@ -74,6 +74,11 @@
           "href": "/cli/rollback",
           "file": "docs/cli/rollback.mdx"
         },
+        {
+          "title": "Validate",
+          "href": "/cli/validate",
+          "file": "docs/cli/validate.mdx"
+        },
         {
           "title": "Create",
           "href": "/cli/create",

pkg/roll/execute.go

@@ -15,6 +15,21 @@ import (
 	"github.com/xataio/pgroll/pkg/schema"
 )
 
+func (m *Roll) Validate(ctx context.Context, migration *migrations.Migration) error {
+	if m.skipValidation {
+		return nil
+	}
+	lastSchema, err := m.state.LastSchema(ctx, m.schema)
+	if err != nil {
+		return err
+	}
+	err = migration.Validate(ctx, lastSchema)
+	if err != nil {
+		return fmt.Errorf("migration '%s' is invalid: %w", migration.Name, err)
+	}
+	return nil
+}
+
 // Start will apply the required changes to enable supporting the new schema version
 func (m *Roll) Start(ctx context.Context, migration *migrations.Migration, cfg *backfill.Config) error {
 	// Fail early if we have existing schema without migration history
@@ -28,6 +43,10 @@ func (m *Roll) Start(ctx context.Context, migration *migrations.Migration, cfg *
 
 	m.logger.LogMigrationStart(migration)
 
+	if err := m.Validate(ctx, migration); err != nil {
+		return err
+	}
+
 	tablesToBackfill, err := m.StartDDLOperations(ctx, migration)
 	if err != nil {
 		return err
@@ -50,22 +69,10 @@ func (m *Roll) StartDDLOperations(ctx context.Context, migration *migrations.Mig
 	}
 
 	// create a new active migration (guaranteed to be unique by constraints)
-	newSchema, err := m.state.Start(ctx, m.schema, migration)
-	if err != nil {
+	if err = m.state.Start(ctx, m.schema, migration); err != nil {
 		return nil, fmt.Errorf("unable to start migration: %w", err)
 	}
 
-	// validate migration
-	if !m.skipValidation {
-		err = migration.Validate(ctx, newSchema)
-		if err != nil {
-			if err := m.state.Rollback(ctx, m.schema, migration.Name); err != nil {
-				fmt.Printf("failed to rollback migration: %s\n", err)
-			}
-			return nil, fmt.Errorf("migration is invalid: %w", err)
-		}
-	}
-
 	// run any BeforeStartDDL hooks
 	if m.migrationHooks.BeforeStartDDL != nil {
 		if err := m.migrationHooks.BeforeStartDDL(m); err != nil {
@@ -89,7 +96,7 @@ func (m *Roll) StartDDLOperations(ctx context.Context, migration *migrations.Mig
 
 	// Reread the latest schema as validation may have updated the schema object
 	// in memory.
-	newSchema, err = m.state.ReadSchema(ctx, m.schema)
+	newSchema, err := m.state.ReadSchema(ctx, m.schema)
 	if err != nil {
 		return nil, fmt.Errorf("unable to read schema: %w", err)
 	}

pkg/state/history_test.go

@@ -55,7 +55,7 @@ func TestSchemaHistoryReturnsFullSchemaHistory(t *testing.T) {
 
 		// Start and complete both migrations
 		for _, mig := range migs {
-			_, err := state.Start(ctx, "public", &mig)
+			err := state.Start(ctx, "public", &mig)
 			require.NoError(t, err)
 			err = state.Complete(ctx, "public", mig.Name)
 			require.NoError(t, err)

pkg/state/state.go

@@ -236,24 +236,29 @@ func (s *State) Status(ctx context.Context, schema string) (*Status, error) {
 // this will effectively activate a new migration period, so `IsActiveMigrationPeriod` will return true
 // until the migration is completed
 // This method will return the current schema (before the migration is applied)
-func (s *State) Start(ctx context.Context, schemaname string, migration *migrations.Migration) (*schema.Schema, error) {
+func (s *State) Start(ctx context.Context, schemaname string, migration *migrations.Migration) error {
 	rawMigration, err := json.Marshal(migration)
 	if err != nil {
-		return nil, fmt.Errorf("unable to marshal migration: %w", err)
+		return fmt.Errorf("unable to marshal migration: %w", err)
 	}
 
 	// create a new migration object and return the previous known schema
 	// if there is no previous migration, read the schema from postgres
-	stmt := fmt.Sprintf(`
-		INSERT INTO %[1]s.migrations (schema, name, parent, migration) VALUES ($1, $2, %[1]s.latest_version($1), $3)
-		RETURNING (
-			SELECT COALESCE(
-				(SELECT resulting_schema FROM %[1]s.migrations WHERE schema=$1 AND name=%[1]s.latest_version($1)),
-				%[1]s.read_schema($1))
-		)`, pq.QuoteIdentifier(s.schema))
+	stmt := fmt.Sprintf(`INSERT INTO %[1]s.migrations (schema, name, parent, migration) VALUES ($1, $2, %[1]s.latest_version($1), $3)`,
+		pq.QuoteIdentifier(s.schema))
+
+	_, err = s.pgConn.ExecContext(ctx, stmt, schemaname, migration.Name, rawMigration)
+	return err
+}
+
+func (s *State) LastSchema(ctx context.Context, schemaname string) (*schema.Schema, error) {
+	stmt := fmt.Sprintf(`SELECT COALESCE(
+		(SELECT resulting_schema FROM %[1]s.migrations WHERE schema=$1 AND name=%[1]s.latest_version($1)),
+		%[1]s.read_schema($1))`,
+		pq.QuoteIdentifier(s.schema))
 
 	var rawSchema string
-	err = s.pgConn.QueryRowContext(ctx, stmt, schemaname, migration.Name, rawMigration).Scan(&rawSchema)
+	err := s.pgConn.QueryRowContext(ctx, stmt, schemaname).Scan(&rawSchema)
 	if err != nil {
 		return nil, err
 	}

pkg/state/state_test.go

@@ -35,8 +35,15 @@ func TestSchemaOptionIsRespected(t *testing.T) {
 			t.Fatal(err)
 		}
 
-		// check that starting a new migration returns the already existing table
-		currentSchema, err := state.Start(ctx, "public", &migrations.Migration{
+		// check that we can retrieve the already existing table
+		currentSchema, err := state.LastSchema(ctx, "public")
+		assert.NoError(t, err)
+
+		assert.Equal(t, 1, len(currentSchema.Tables))
+		assert.Equal(t, "public", currentSchema.Name)
+
+		// check that we can start the migration
+		err = state.Start(ctx, "public", &migrations.Migration{
 			Name: "1_add_column",
 			Operations: migrations.Operations{
 				&migrations.OpAddColumn{
@@ -49,9 +56,6 @@ func TestSchemaOptionIsRespected(t *testing.T) {
 			},
 		})
 		assert.NoError(t, err)
-
-		assert.Equal(t, 1, len(currentSchema.Tables))
-		assert.Equal(t, "public", currentSchema.Name)
 	})
 }