Allow migration files to specify a versionSchema field (#884)

#852 removed support for specifying a migration name via the `name`
field in a migration file. While having an unambiguous source of truth
for the migration name (the filename) is good, it is still often
desirable to decouple the filename from the name of the version schema
that the migration will create when applied because filenames:

* can be longer than the limit Postgres imposes on schema names
* can contain characters that are not legal in Postgres schema names
* must be named so that migration files are ordered lexicographically on
disk

This PR adds support for a new `version_schema` field in migration files
that allows the migration author to specify the name of the version
schema that the migration will create.

## Example

```yaml
# migration files can now specify the version schema 
# name to be created by the migration
version_schema: my_version_schema
operations:
  - create_table:
      name: items
      columns:
        - name: id
          type: serial
          pk: true
        - name: name
          type: varchar(255)
```

Running this migration:

```
$ pgroll start migrations/01_create_table.yaml --complete
```

Creates this version schema:

```
+-----------------------+-------------------+
| Name                  | Owner             |
|-----------------------+-------------------|
...
| public_my_version_schema | postgres       |
+-----------------------+-------------------+
```

Had the `version_schema` field not been specified in the migration file
the version schema would have been:

```
+-----------------------+-------------------+
| Name                  | Owner             |
|-----------------------+-------------------|
...
| public_01_create_table | postgres         |
+-----------------------+-------------------+
```

---

This PR is part of a stack:
* #884 (this PR)
* #886
* #887

Part of #882

Author: andrew-farries

cmd/migrate.go

@@ -36,7 +36,7 @@ func migrateCmd() *cobra.Command {
 			}
 			defer m.Close()
 
-			latestVersion, err := m.State().LatestVersion(ctx, m.Schema())
+			latestMigration, err := m.State().LatestMigration(ctx, m.Schema())
 			if err != nil {
 				return fmt.Errorf("unable to determine latest version: %w", err)
 			}
@@ -46,7 +46,7 @@ func migrateCmd() *cobra.Command {
 				return fmt.Errorf("unable to determine active migration period: %w", err)
 			}
 			if active {
-				fmt.Printf("migration %q is active\n", *latestVersion)
+				fmt.Printf("migration %q is active\n", *latestMigration)
 				return nil
 			}
 

cmd/start.go

@@ -105,8 +105,7 @@ func runMigration(ctx context.Context, m *roll.Roll, migration *migrations.Migra
 		}
 	}
 
-	version := migration.Name
-	viewName := roll.VersionedSchemaName(flags.Schema(), version)
+	viewName := roll.VersionedSchemaName(flags.Schema(), migration.VersionSchemaName())
 	msg := fmt.Sprintf("New version of the schema available under the postgres %q schema", viewName)
 	sp.Success(msg)
 

pkg/migrations/migrations.go

@@ -68,15 +68,26 @@ type RequiresSchemaRefreshOperation interface {
 type (
 	Operations []Operation
 	Migration  struct {
-		Name       string     `json:"-"`
-		Operations Operations `json:"operations"`
+		Name          string     `json:"-"`
+		VersionSchema string     `json:"version_schema,omitempty"`
+		Operations    Operations `json:"operations"`
 	}
 	RawMigration struct {
-		Name       string          `json:"-"`
-		Operations json.RawMessage `json:"operations"`
+		Name          string          `json:"-"`
+		VersionSchema string          `json:"version_schema,omitempty"`
+		Operations    json.RawMessage `json:"operations"`
 	}
 )
 
+// VersionSchemaName returns the version schema name for the migration.
+// It defaults to the migration name if no version schema is set.
+func (m *Migration) VersionSchemaName() string {
+	if m.VersionSchema != "" {
+		return m.VersionSchema
+	}
+	return m.Name
+}
+
 // Validate will check that the migration can be applied to the given schema
 // returns a descriptive error if the migration is invalid
 func (m *Migration) Validate(ctx context.Context, s *schema.Schema) error {

pkg/migrations/op_common.go

@@ -132,8 +132,9 @@ func ParseMigration(raw *RawMigration) (*Migration, error) {
 	}
 
 	return &Migration{
-		Name:       raw.Name,
-		Operations: ops,
+		Name:          raw.Name,
+		VersionSchema: raw.VersionSchema,
+		Operations:    ops,
 	}, nil
 }
 

pkg/roll/execute.go

@@ -136,16 +136,15 @@ func (m *Roll) StartDDLOperations(ctx context.Context, migration *migrations.Mig
 	}
 
 	// create views for the new version
-	if err := m.ensureViews(ctx, newSchema, migration.Name); err != nil {
+	if err := m.ensureViews(ctx, newSchema, migration); err != nil {
 		return nil, err
 	}
 
 	return tablesToBackfill, nil
 }
 
-func (m *Roll) ensureViews(ctx context.Context, schema *schema.Schema, version string) error {
-	// create schema for the new version
-	versionSchema := VersionedSchemaName(m.schema, version)
+func (m *Roll) ensureViews(ctx context.Context, schema *schema.Schema, mig *migrations.Migration) error {
+	versionSchema := VersionedSchemaName(m.schema, mig.VersionSchemaName())
 	_, err := m.pgConn.ExecContext(ctx, fmt.Sprintf("CREATE SCHEMA IF NOT EXISTS %s", pq.QuoteIdentifier(versionSchema)))
 	if err != nil {
 		return err
@@ -156,13 +155,13 @@ func (m *Roll) ensureViews(ctx context.Context, schema *schema.Schema, version s
 		if table.Deleted {
 			continue
 		}
-		err = m.ensureView(ctx, version, name, table)
+		err = m.ensureView(ctx, mig.VersionSchemaName(), name, table)
 		if err != nil {
 			return fmt.Errorf("unable to create view: %w", err)
 		}
 	}
 
-	m.logger.LogSchemaCreation(version, versionSchema)
+	m.logger.LogSchemaCreation(mig.VersionSchemaName(), versionSchema)
 
 	return nil
 }
@@ -237,7 +236,7 @@ func (m *Roll) Complete(ctx context.Context) error {
 			return fmt.Errorf("unable to read schema: %w", err)
 		}
 
-		err = m.ensureViews(ctx, currentSchema, migration.Name)
+		err = m.ensureViews(ctx, currentSchema, migration)
 		if err != nil {
 			return err
 		}
@@ -266,7 +265,7 @@ func (m *Roll) Rollback(ctx context.Context) error {
 
 	if !m.disableVersionSchemas {
 		// delete the schema and view for the new version
-		versionSchema := VersionedSchemaName(m.schema, migration.Name)
+		versionSchema := VersionedSchemaName(m.schema, migration.VersionSchemaName())
 		_, err = m.pgConn.ExecContext(ctx, fmt.Sprintf("DROP SCHEMA IF EXISTS %s CASCADE", pq.QuoteIdentifier(versionSchema)))
 		if err != nil {
 			return err
@@ -276,15 +275,15 @@ func (m *Roll) Rollback(ctx context.Context) error {
 	}
 
 	// get the name of the previous version of the schema
-	previousVersion, err := m.state.PreviousVersion(ctx, m.schema, true)
+	previousMigration, err := m.state.PreviousMigration(ctx, m.schema, true)
 	if err != nil {
 		return fmt.Errorf("unable to get name of previous version: %w", err)
 	}
 
 	// get the schema after the previous migration was applied
 	schema := schema.New()
-	if previousVersion != nil {
-		schema, err = m.state.SchemaAfterMigration(ctx, m.schema, *previousVersion)
+	if previousMigration != nil {
+		schema, err = m.state.SchemaAfterMigration(ctx, m.schema, *previousMigration)
 		if err != nil {
 			return fmt.Errorf("unable to read schema: %w", err)
 		}

pkg/roll/execute_test.go

@@ -155,6 +155,49 @@ func TestPreviousVersionIsDroppedAfterMigrationCompletion(t *testing.T) {
 			}
 		})
 	})
+
+	t.Run("when the previous version sets a non-default version schema name", func(t *testing.T) {
+		testutils.WithMigratorAndConnectionToContainer(t, func(m *roll.Roll, db *sql.DB) {
+			ctx := context.Background()
+
+			migs := []migrations.Migration{
+				{
+					Name:          "01_create_table",
+					VersionSchema: "01_foo",
+					Operations: migrations.Operations{
+						&migrations.OpCreateTable{
+							Name: "users",
+							Columns: []migrations.Column{
+								{Name: "id", Type: "serial", Pk: true},
+							},
+						},
+					},
+				},
+				{
+					Name: "02_create_another_table",
+					Operations: migrations.Operations{
+						&migrations.OpCreateTable{
+							Name: "items",
+							Columns: []migrations.Column{
+								{Name: "id", Type: "serial", Pk: true},
+							},
+						},
+					},
+				},
+			}
+
+			// Start and complete both migrations
+			for _, mig := range migs {
+				err := m.Start(ctx, &mig, backfill.NewConfig())
+				require.NoError(t, err)
+				err = m.Complete(ctx)
+				require.NoError(t, err)
+			}
+
+			// Ensure that the schema for the first migration has been dropped
+			require.False(t, schemaExists(t, db, roll.VersionedSchemaName("public", "01_foo")))
+		})
+	})
 }
 
 func TestNoVersionSchemaForRawSQLMigrationsOptionIsRespected(t *testing.T) {
@@ -215,23 +258,48 @@ func TestNoVersionSchemaForRawSQLMigrationsOptionIsRespected(t *testing.T) {
 func TestSchemaIsDroppedAfterMigrationRollback(t *testing.T) {
 	t.Parallel()
 
-	testutils.WithMigratorAndConnectionToContainer(t, func(mig *roll.Roll, db *sql.DB) {
-		ctx := context.Background()
-		version := "1_create_table"
+	t.Run("when the migration does not set an explicit version schema name ", func(t *testing.T) {
+		testutils.WithMigratorAndConnectionToContainer(t, func(mig *roll.Roll, db *sql.DB) {
+			ctx := context.Background()
+			version := "1_create_table"
 
-		if err := mig.Start(ctx, &migrations.Migration{Name: version, Operations: migrations.Operations{createTableOp("table1")}}, backfill.NewConfig()); err != nil {
-			t.Fatalf("Failed to start migration: %v", err)
-		}
-		if err := mig.Rollback(ctx); err != nil {
-			t.Fatalf("Failed to rollback migration: %v", err)
-		}
+			if err := mig.Start(ctx, &migrations.Migration{
+				Name:       version,
+				Operations: migrations.Operations{createTableOp("table1")},
+			}, backfill.NewConfig()); err != nil {
+				t.Fatalf("Failed to start migration: %v", err)
+			}
+			if err := mig.Rollback(ctx); err != nil {
+				t.Fatalf("Failed to rollback migration: %v", err)
+			}
 
-		//
-		// Check that the schema has been dropped
-		//
-		if schemaExists(t, db, roll.VersionedSchemaName(cSchema, version)) {
-			t.Errorf("Expected schema %q to not exist", version)
-		}
+			// Check that the schema has been dropped
+			if schemaExists(t, db, roll.VersionedSchemaName(cSchema, version)) {
+				t.Errorf("Expected schema %q to not exist", version)
+			}
+		})
+	})
+
+	t.Run("when the migration does set an explicit version schema name ", func(t *testing.T) {
+		testutils.WithMigratorAndConnectionToContainer(t, func(mig *roll.Roll, db *sql.DB) {
+			ctx := context.Background()
+
+			if err := mig.Start(ctx, &migrations.Migration{
+				Name:          "1_create_table",
+				VersionSchema: "1_foo",
+				Operations:    migrations.Operations{createTableOp("table1")},
+			}, backfill.NewConfig()); err != nil {
+				t.Fatalf("Failed to start migration: %v", err)
+			}
+			if err := mig.Rollback(ctx); err != nil {
+				t.Fatalf("Failed to rollback migration: %v", err)
+			}
+
+			// Check that the schema has been dropped
+			if schemaExists(t, db, roll.VersionedSchemaName(cSchema, "1_foo")) {
+				t.Errorf("Expected schema %q to not exist", "1_foo")
+			}
+		})
 	})
 }
 
@@ -710,6 +778,43 @@ func TestRollSchemaMethodReturnsCorrectSchema(t *testing.T) {
 	})
 }
 
+func TestLatestVersionAndLatestMigrationMethodsRespectVersionSchemaAndName(t *testing.T) {
+	t.Parallel()
+
+	testutils.WithMigratorAndConnectionToContainer(t, func(r *roll.Roll, db *sql.DB) {
+		ctx := context.Background()
+
+		// Create a migration with an explicit version schema
+		mig := &migrations.Migration{
+			Name:          "01_create_table",
+			VersionSchema: "01_foo",
+			Operations:    migrations.Operations{createTableOp("table1")},
+		}
+
+		// Start and complete a migration
+		err := r.Start(ctx, mig, backfill.NewConfig())
+		require.NoError(t, err)
+		err = r.Complete(ctx)
+		require.NoError(t, err)
+
+		// Get the latest version
+		latestVersion, err := r.State().LatestVersion(ctx, "public")
+		require.NoError(t, err)
+
+		// Get the latest migration name
+		latestMigration, err := r.State().LatestMigration(ctx, "public")
+		require.NoError(t, err)
+
+		// Assert that the latest version is correct
+		require.NotNil(t, latestVersion)
+		require.Equal(t, "01_foo", *latestVersion)
+
+		// Assert that the latest migration name is correct
+		require.NotNil(t, latestMigration)
+		require.Equal(t, "01_create_table", *latestMigration)
+	})
+}
+
 func TestWithSearchPathOptionIsRespected(t *testing.T) {
 	t.Parallel()