Fix previous_version: Take inferred migrations into account (#631)

Add `includeInferred` parameter to function `previous_version`.
If it's `false`, it will behave exactly as it did before this PR. It
will be called with `false` in `Complete`, to ignore inferred migrations
when dropping the version schema, because inferred migrations do not
have version schemas.
If it's `true`, `inferred` migrations will be taken into account when
finding the previous version. It will be called with `true` when finding
the previous version for `Rollback`.

When inferred migrations were not taken into account, `Rollback` was
causing a segmentation violation because of nil pointer for new
columns/tables added by SQL.

This PR also modifies the test
`TestNoVersionSchemaForRawSQLMigrationsOptionIsRespected` so that we
will be testing `PreviousVersion` with both options.

Fixes: #623

Author: agedemenli

pkg/roll/execute.go

@@ -155,7 +155,7 @@ func (m *Roll) Complete(ctx context.Context) error {
 
 	// Drop the old schema
 	if !m.disableVersionSchemas && (!migration.ContainsRawSQLOperation() || !m.noVersionSchemaForRawSQL) {
-		prevVersion, err := m.state.PreviousVersion(ctx, m.schema)
+		prevVersion, err := m.state.PreviousVersion(ctx, m.schema, false)
 		if err != nil {
 			return fmt.Errorf("unable to get name of previous version: %w", err)
 		}
@@ -241,7 +241,7 @@ 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)
+	previousVersion, err := m.state.PreviousVersion(ctx, m.schema, true)
 	if err != nil {
 		return fmt.Errorf("unable to get name of previous version: %w", err)
 	}

pkg/roll/execute_test.go

@@ -185,13 +185,20 @@ func TestNoVersionSchemaForRawSQLMigrationsOptionIsRespected(t *testing.T) {
 		err = mig.Start(ctx, &migrations.Migration{Name: "03_create_table", Operations: migrations.Operations{createTableOp("table3")}})
 		require.NoError(t, err)
 
-		// The previous version is migration 01 because there is no version schema
-		// for migration 02 due to the `WithNoVersionSchemaForRawSQL` option
-		prevVersion, err := st.PreviousVersion(ctx, "public")
+		// The previous version is migration 01 if raw SQL migrations are ignored
+		prevVersion, err := st.PreviousVersion(ctx, "public", false)
 		require.NoError(t, err)
 		require.NotNil(t, prevVersion)
 		assert.Equal(t, "01_create_table", *prevVersion)
 
+		// The previous version is migration 02 (inferred) but there is no version schema
+		// for migration 02 due to the `WithNoVersionSchemaForRawSQL` option
+		prevVersion, err = st.PreviousVersion(ctx, "public", true)
+		require.NoError(t, err)
+		require.NotNil(t, prevVersion)
+		assert.Equal(t, "02_create_table", *prevVersion)
+		assert.False(t, schemaExists(t, db, roll.VersionedSchemaName(schema, "02_create_table")))
+
 		// Complete the third migration
 		err = mig.Complete(ctx)
 		require.NoError(t, err)

pkg/state/init.sql

@@ -76,7 +76,7 @@ STABLE;
 -- Get the name of the previous version of the schema, or NULL if there is none.
 -- This ignores previous versions for which no version schema exists, such as
 -- versions corresponding to inferred migrations.
-CREATE OR REPLACE FUNCTION placeholder.previous_version (schemaname name)
+CREATE OR REPLACE FUNCTION placeholder.previous_version (schemaname name, includeInferred boolean)
     RETURNS text
     AS $$
     WITH RECURSIVE ancestors AS (
@@ -107,10 +107,17 @@ CREATE OR REPLACE FUNCTION placeholder.previous_version (schemaname name)
             a.name
         FROM
             ancestors a
-        JOIN information_schema.schemata s ON s.schema_name = schemaname || '_' || a.name
     WHERE
-        migration_type = 'pgroll'
-        AND a.depth > 0
+        a.depth > 0
+        AND (includeInferred
+            OR (a.migration_type = 'pgroll'
+                AND EXISTS (
+                    SELECT
+                        s.schema_name
+                    FROM
+                        information_schema.schemata s
+                    WHERE
+                        s.schema_name = schemaname || '_' || a.name)))
     ORDER BY
         a.depth ASC
     LIMIT 1;

pkg/state/state.go

@@ -136,11 +136,11 @@ func (s *State) LatestVersion(ctx context.Context, schema string) (*string, erro
 }
 
 // PreviousVersion returns the name of the previous version schema
-func (s *State) PreviousVersion(ctx context.Context, schema string) (*string, error) {
+func (s *State) PreviousVersion(ctx context.Context, schema string, includeInferred bool) (*string, error) {
 	var parent *string
 	err := s.pgConn.QueryRowContext(ctx,
-		fmt.Sprintf("SELECT %s.previous_version($1)", pq.QuoteIdentifier(s.schema)),
-		schema).Scan(&parent)
+		fmt.Sprintf("SELECT %s.previous_version($1, $2)", pq.QuoteIdentifier(s.schema)),
+		schema, includeInferred).Scan(&parent)
 	if err != nil {
 		return nil, err
 	}