[baseline] Make pgroll pull pull only migrations after the most recent baseline (#833)

Make the `pgroll pull` command pull only migrations after the most
recent baseline. A baseline migration defines a new starting point for
the schema history, and as such, `pgroll pull` should not consider
migrations prior to the most recent baseline.

The change here is to the `State.SchemaHistory` method, which now only
returns migrations after the most recent baseline.

Part of #364

Author: andrew-farries

pkg/roll/missing_test.go

@@ -211,6 +211,61 @@ func TestMissingMigrations(t *testing.T) {
 			require.Len(t, migs, 0)
 		})
 	})
+
+	t.Run("baseline migrations are not considered missing", func(t *testing.T) {
+		fs := fstest.MapFS{}
+
+		testutils.WithMigratorAndConnectionToContainer(t, func(m *roll.Roll, _ *sql.DB) {
+			ctx := context.Background()
+
+			// Create a baseline migration
+			err := m.CreateBaseline(ctx, "01_initial_version")
+			require.NoError(t, err)
+
+			// Get missing migrations
+			migs, err := m.MissingMigrations(ctx, fs)
+			require.NoError(t, err)
+
+			// Assert that there are no migrations missing
+			require.Len(t, migs, 0)
+		})
+	})
+
+	t.Run("migrations before a baseline migration are not considered missing", func(t *testing.T) {
+		fs := fstest.MapFS{}
+
+		testutils.WithMigratorAndConnectionToContainer(t, func(m *roll.Roll, _ *sql.DB) {
+			ctx := context.Background()
+
+			// Apply a migration to the target database
+			err := m.Start(ctx, exampleMig(t, "01_migration_1"), backfill.NewConfig())
+			require.NoError(t, err)
+			err = m.Complete(ctx)
+			require.NoError(t, err)
+
+			// Create a baseline migration
+			err = m.CreateBaseline(ctx, "01_initial_version")
+			require.NoError(t, err)
+
+			// Apply another migration to the target database
+			err = m.Start(ctx, exampleMig(t, "02_migration_2"), backfill.NewConfig())
+			require.NoError(t, err)
+			err = m.Complete(ctx)
+			require.NoError(t, err)
+
+			// Get missing migrations
+			migs, err := m.MissingMigrations(ctx, fs)
+			require.NoError(t, err)
+
+			// Assert that there is one migration missing from the local directory
+			require.Len(t, migs, 1)
+
+			// Assert that the missing migration is the one that was applied after the baseline
+			mig, err := migrations.ParseMigration(migs[0])
+			require.NoError(t, err)
+			require.Equal(t, exampleMig(t, "02_migration_2"), mig)
+		})
+	})
 }
 
 func TestMissingMigrationsWithOldMigrationFormats(t *testing.T) {

pkg/state/history.go

@@ -24,8 +24,15 @@ type HistoryEntry struct {
 func (s *State) SchemaHistory(ctx context.Context, schema string) ([]HistoryEntry, error) {
 	rows, err := s.pgConn.QueryContext(ctx,
 		fmt.Sprintf(`SELECT name, migration, created_at
-			FROM %s.migrations
-			WHERE schema=$1 ORDER BY created_at`,
+			FROM %[1]s.migrations
+			WHERE schema=$1
+			AND created_at > COALESCE(
+			(
+				SELECT MAX(created_at) FROM %[1]s.migrations
+				WHERE schema = $1 AND migration_type = 'baseline'
+			),
+			'-infinity'::timestamptz
+			) ORDER BY created_at`,
 			pq.QuoteIdentifier(s.schema)), schema)
 	if err != nil {
 		return nil, err

pkg/state/history_test.go

@@ -79,6 +79,85 @@ func TestSchemaHistoryReturnsFullSchemaHistory(t *testing.T) {
 	})
 }
 
+func TestSchemaHistoryDoesNotReturnBaselineMigrations(t *testing.T) {
+	t.Parallel()
+
+	t.Run("baseline migration does not appear in schema history", func(t *testing.T) {
+		testutils.WithStateAndConnectionToContainer(t, func(state *state.State, db *sql.DB) {
+			ctx := context.Background()
+
+			// Create a baseline migration
+			err := state.CreateBaseline(ctx, "public", "01_initial_version")
+			require.NoError(t, err)
+
+			// Get the schema history
+			res, err := state.SchemaHistory(ctx, "public")
+			require.NoError(t, err)
+
+			// Assert that the schema history is empty
+			assert.Equal(t, 0, len(res))
+		})
+	})
+
+	t.Run("migrations before the most recent baseline do not appear in the schema history", func(t *testing.T) {
+		testutils.WithStateAndConnectionToContainer(t, func(state *state.State, db *sql.DB) {
+			ctx := context.Background()
+
+			// Execute DDL to create an inferred migration
+			_, err := db.ExecContext(ctx, "CREATE TABLE users (id int)")
+			require.NoError(t, err)
+
+			// Create a baseline migration
+			err = state.CreateBaseline(ctx, "public", "01_initial_version")
+			require.NoError(t, err)
+
+			// Get the schema history
+			res, err := state.SchemaHistory(ctx, "public")
+			require.NoError(t, err)
+
+			// Assert that the schema history is empty
+			assert.Equal(t, 0, len(res))
+		})
+	})
+
+	t.Run("migrations after the most recent baseline are included in the history", func(t *testing.T) {
+		testutils.WithStateAndConnectionToContainer(t, func(state *state.State, db *sql.DB) {
+			ctx := context.Background()
+
+			// Execute DDL to create an inferred migration
+			_, err := db.ExecContext(ctx, "CREATE TABLE users (id int)")
+			require.NoError(t, err)
+
+			// Create a baseline migration
+			err = state.CreateBaseline(ctx, "public", "01_initial_version")
+			require.NoError(t, err)
+
+			// Execute DDL to create another inferred migration
+			_, err = db.ExecContext(ctx, "CREATE TABLE fruits (id int)")
+			require.NoError(t, err)
+
+			// Get the schema history
+			res, err := state.SchemaHistory(ctx, "public")
+			require.NoError(t, err)
+
+			// Assert that one migration is present in the schema history
+			require.Equal(t, 1, len(res))
+
+			// Deserialize the migration from the history.
+			mig, err := migrations.ParseMigration(&res[0].Migration)
+			require.NoError(t, err)
+
+			// Assert that the migration is the one that was created after the baseline
+			expectedOperations := migrations.Operations{
+				&migrations.OpRawSQL{
+					Up: "CREATE TABLE fruits (id int)",
+				},
+			}
+			assert.Equal(t, expectedOperations, mig.Operations)
+		})
+	})
+}
+
 func ptr[T any](v T) *T {
 	return &v
 }