Make pgroll latest schema ignore migrations for which no version schema exists (#916)

Fix the problem described in #915 where `pgroll latest schema` will
return version schema names for which no version schema exists in the
database, such as for `inferred` migrations.

---

1. Apply a migration
```
pgroll  start examples/01_create_tables.yaml --complete
```

2. Create an inferred migration
```
CREATE TABLE foo(id int PRIMARY KEY)
```

3. Run `pgroll latest schema`

Before this PR the output would be incorrect:

```
public_sql_fca0901e
```

The most recent migration (inferred from the `CREATE TABLE`) did not
create a version schema - the most recent schema version is still
`public_01_create_tables`.

This PR corrects the behaviour of `pgroll latest schema` to consider
only migrations for which a version schema exists. So the output is:

```
public_01_create_tables
```

as expected.

---


Fixes #915 

Related to #872

Author: andrew-farries

pkg/roll/execute.go

@@ -175,7 +175,7 @@ func (m *Roll) Complete(ctx context.Context) error {
 
 	// Drop the old schema
 	if !m.disableVersionSchemas {
-		prevVersion, err := m.state.PreviousVersion(ctx, m.schema, false)
+		prevVersion, err := m.state.PreviousVersion(ctx, m.schema)
 		if err != nil {
 			return fmt.Errorf("unable to get name of previous version: %w", err)
 		}

pkg/roll/latest_test.go

@@ -72,7 +72,37 @@ func TestLatestVersionRemote(t *testing.T) {
 			latestVersion, err := m.LatestVersionRemote(ctx)
 			require.NoError(t, err)
 
-			// Assert latest migration name
+			// Assert latest migration version schema name
+			assert.Equal(t, "01_foo", latestVersion)
+		})
+	})
+
+	t.Run("inferred migrations without a version schema are ignored", func(t *testing.T) {
+		testutils.WithMigratorAndConnectionToContainer(t, func(m *roll.Roll, db *sql.DB) {
+			ctx := context.Background()
+
+			// Start and complete a migration
+			err := m.Start(ctx, &migrations.Migration{
+				Name:          "01_first_migration",
+				VersionSchema: "01_foo",
+				Operations: migrations.Operations{
+					&migrations.OpRawSQL{Up: "SELECT 1"},
+				},
+			}, backfill.NewConfig())
+			require.NoError(t, err)
+			err = m.Complete(ctx)
+			require.NoError(t, err)
+
+			// Run some DDL to generate an inferred migration
+			_, err = db.ExecContext(ctx, "CREATE TABLE apples(id int)")
+			require.NoError(t, err)
+
+			// Get the latest version in the target schema
+			latestVersion, err := m.LatestVersionRemote(ctx)
+			require.NoError(t, err)
+
+			// Assert latest migration version schema name; the inferred migration
+			// name is ignored
 			assert.Equal(t, "01_foo", latestVersion)
 		})
 	})

pkg/state/init.sql

@@ -68,31 +68,7 @@ $$
 LANGUAGE SQL
 STABLE;
 
--- Get the latest version schema name
-CREATE OR REPLACE FUNCTION placeholder.latest_version (schemaname name)
-    RETURNS text
-    SECURITY DEFINER
-    SET search_path = placeholder, pg_catalog, pg_temp
-    AS $$
-    SELECT
-        COALESCE(p.migration ->> 'version_schema', p.name)
-    FROM
-        placeholder.migrations p
-    WHERE
-        NOT EXISTS (
-            SELECT
-                1
-            FROM
-                placeholder.migrations c
-            WHERE
-                SCHEMA = schemaname
-                AND c.parent = p.name)
-        AND SCHEMA = schemaname
-$$
-LANGUAGE SQL
-STABLE;
-
--- Get the name of the latest version schema, or NULL if there is none.
+-- Get the name of the latest migration, or NULL if there is none.
 -- This will be the same as the version-schema name of the migration in most
 -- cases, unless the migration sets its `versionSchema` field.
 CREATE OR REPLACE FUNCTION placeholder.latest_migration (schemaname name)
@@ -118,16 +94,13 @@ $$
 LANGUAGE SQL
 STABLE;
 
--- Get the previous version schema name, 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, includeInferred boolean)
+-- Get the name of the previous migration, or NULL if there is none.
+CREATE OR REPLACE FUNCTION placeholder.previous_migration (schemaname name)
     RETURNS text
     AS $$
     WITH RECURSIVE ancestors AS (
         SELECT
             name,
-            migration ->> 'version_schema' AS versionSchema,
             schema,
             parent,
             migration_type,
@@ -140,7 +113,6 @@ CREATE OR REPLACE FUNCTION placeholder.previous_version (schemaname name, includ
         UNION ALL
         SELECT
             m.name,
-            m.migration ->> 'version_schema' AS versionSchema,
             m.schema,
             m.parent,
             m.migration_type,
@@ -151,68 +123,92 @@ CREATE OR REPLACE FUNCTION placeholder.previous_version (schemaname name, includ
                 AND m.schema = a.schema
 )
         SELECT
-            COALESCE(a.versionSchema, a.name)
+            a.name
         FROM
             ancestors a
     WHERE
         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 || '_' || COALESCE(a.versionSchema, a.name))))
     ORDER BY
         a.depth ASC
     LIMIT 1;
 $$
 LANGUAGE SQL
 STABLE;
 
--- Get the name of the previous migration, or NULL if there is none.
-CREATE OR REPLACE FUNCTION placeholder.previous_migration (schemaname name)
+-- find_version_schema finds a recent version schema for a given schema name.
+-- How recent is determined by the minDepth parameter: for a minDepth of 0, it
+-- returns the latest version schema, for a minDepth of 1, it returns the
+-- previous version schema, and so on.
+-- Only version schemas that exist in the database are considered; migrations
+-- without version schema (such as inferred migrations) are ignored.
+CREATE OR REPLACE FUNCTION find_version_schema (p_schema_name name, p_depth integer DEFAULT 0)
     RETURNS text
     AS $$
     WITH RECURSIVE ancestors AS (
         SELECT
             name,
+            COALESCE(migration ->> 'version_schema', name) AS version_schema,
             schema,
             parent,
-            migration_type,
             0 AS depth
         FROM
             placeholder.migrations
         WHERE
-            name = placeholder.latest_migration (schemaname)
-            AND SCHEMA = schemaname
+            name = placeholder.latest_migration (p_schema_name)
+            AND SCHEMA = p_schema_name
         UNION ALL
         SELECT
             m.name,
+            COALESCE(m.migration ->> 'version_schema', m.name) AS version_schema,
             m.schema,
             m.parent,
-            m.migration_type,
             a.depth + 1
         FROM
             placeholder.migrations m
             JOIN ancestors a ON m.name = a.parent
                 AND m.schema = a.schema
 )
         SELECT
-            a.name
+            a.version_schema
         FROM
             ancestors a
     WHERE
-        a.depth > 0
+        EXISTS (
+            SELECT
+                1
+            FROM
+                information_schema.schemata s
+            WHERE
+                s.schema_name = p_schema_name || '_' || a.version_schema)
     ORDER BY
-        a.depth ASC
+        a.depth ASC OFFSET p_depth
     LIMIT 1;
 $$
 LANGUAGE SQL
 STABLE;
 
+-- previous_version returns the name of the previous version schema for a given
+-- schema name or NULL if there is no previous version schema.
+CREATE OR REPLACE FUNCTION previous_version (schemaname name)
+    RETURNS text
+    AS $$
+    SELECT
+        placeholder.find_version_schema (schemaname, 1);
+$$
+LANGUAGE SQL
+STABLE;
+
+-- latest_version returns the name of the latest version schema for a given
+-- schema name or NULL if there are no version schema.
+CREATE OR REPLACE FUNCTION latest_version (schemaname name)
+    RETURNS text
+    AS $$
+    SELECT
+        placeholder.find_version_schema (schemaname, 0);
+$$
+LANGUAGE SQL
+STABLE;
+
 -- Get the JSON representation of the current schema
 CREATE OR REPLACE FUNCTION placeholder.read_schema (schemaname text)
     RETURNS jsonb

pkg/state/latest.go

@@ -0,0 +1,65 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package state
+
+import (
+	"context"
+	"fmt"
+
+	"github.com/lib/pq"
+)
+
+// LatestVersion returns the name of the latest version schema, or nil if there
+// is none.
+func (s *State) LatestVersion(ctx context.Context, schema string) (*string, error) {
+	var version *string
+	err := s.pgConn.QueryRowContext(ctx,
+		fmt.Sprintf("SELECT %s.latest_version($1)", pq.QuoteIdentifier(s.schema)),
+		schema).Scan(&version)
+	if err != nil {
+		return nil, err
+	}
+
+	return version, nil
+}
+
+// PreviousVersion returns the name of the previous version schema
+func (s *State) PreviousVersion(ctx context.Context, schema string) (*string, error) {
+	var parent *string
+	err := s.pgConn.QueryRowContext(ctx,
+		fmt.Sprintf("SELECT %s.previous_version($1)", pq.QuoteIdentifier(s.schema)),
+		schema).Scan(&parent)
+	if err != nil {
+		return nil, err
+	}
+
+	return parent, nil
+}
+
+// LatestMigration returns the name of the latest migration, or nil if there
+// is none.
+func (s *State) LatestMigration(ctx context.Context, schema string) (*string, error) {
+	var migration *string
+	err := s.pgConn.QueryRowContext(ctx,
+		fmt.Sprintf("SELECT %s.latest_migration($1)", pq.QuoteIdentifier(s.schema)),
+		schema).Scan(&migration)
+	if err != nil {
+		return nil, err
+	}
+
+	return migration, nil
+}
+
+// PreviousMigration returns the name of the previous migration, or nil if
+// there is none.
+func (s *State) PreviousMigration(ctx context.Context, schema string) (*string, error) {
+	var parent *string
+	err := s.pgConn.QueryRowContext(ctx,
+		fmt.Sprintf("SELECT %s.previous_migration($1)", pq.QuoteIdentifier(s.schema)),
+		schema).Scan(&parent)
+	if err != nil {
+		return nil, err
+	}
+
+	return parent, nil
+}