Implement the drop table operation (#45)

Add support for **drop table** migrations.

* On starting the migration the table is not dropped; but is not present
in the new version of the schema.
* The table is dropped on completion of the migration.
* Rollback is a no-op.

examples/07_drop_table.json

@@ -0,0 +1,10 @@
+{
+  "name": "07_drop_table",
+  "operations": [
+    {
+      "drop_table": {
+        "name": "products"
+      }
+    }
+  ]
+}

pkg/migrations/op_common.go

@@ -13,6 +13,7 @@ type OpName string
 const (
 	OpNameCreateTable OpName = "create_table"
 	OpNameRenameTable OpName = "rename_table"
+	OpNameDropTable   OpName = "drop_table"
 	OpNameAddColumn   OpName = "add_column"
 )
 
@@ -75,6 +76,9 @@ func (v *Operations) UnmarshalJSON(data []byte) error {
 		case OpNameRenameTable:
 			item = &OpRenameTable{}
 
+		case OpNameDropTable:
+			item = &OpDropTable{}
+
 		case OpNameAddColumn:
 			item = &OpAddColumn{}
 
@@ -116,6 +120,9 @@ func (v Operations) MarshalJSON() ([]byte, error) {
 		case *OpRenameTable:
 			opName = OpNameRenameTable
 
+		case *OpDropTable:
+			opName = OpNameDropTable
+
 		case *OpAddColumn:
 			opName = OpNameAddColumn
 

pkg/migrations/op_drop_table.go

@@ -0,0 +1,41 @@
+package migrations
+
+import (
+	"context"
+	"database/sql"
+	"fmt"
+
+	"github.com/lib/pq"
+
+	"pg-roll/pkg/schema"
+)
+
+type OpDropTable struct {
+	Name string `json:"name"`
+}
+
+var _ Operation = (*OpDropTable)(nil)
+
+func (o *OpDropTable) Start(ctx context.Context, conn *sql.DB, schemaName, stateSchema string, s *schema.Schema) error {
+	s.RemoveTable(o.Name)
+	return nil
+}
+
+func (o *OpDropTable) Complete(ctx context.Context, conn *sql.DB) error {
+	_, err := conn.ExecContext(ctx, fmt.Sprintf("DROP TABLE IF EXISTS %s", pq.QuoteIdentifier(o.Name)))
+
+	return err
+}
+
+func (o *OpDropTable) Rollback(ctx context.Context, conn *sql.DB) error {
+	return nil
+}
+
+func (o *OpDropTable) Validate(ctx context.Context, s *schema.Schema) error {
+	table := s.GetTable(o.Name)
+
+	if table == nil {
+		return TableDoesNotExistError{Name: o.Name}
+	}
+	return nil
+}

pkg/migrations/op_drop_table_test.go

@@ -0,0 +1,60 @@
+package migrations_test
+
+import (
+	"database/sql"
+	"testing"
+
+	"pg-roll/pkg/migrations"
+)
+
+func TestDropTable(t *testing.T) {
+	t.Parallel()
+
+	ExecuteTests(t, TestCases{TestCase{
+		name: "drop table",
+		migrations: []migrations.Migration{
+			{
+				Name: "01_create_table",
+				Operations: migrations.Operations{
+					&migrations.OpCreateTable{
+						Name: "users",
+						Columns: []migrations.Column{
+							{
+								Name:       "id",
+								Type:       "serial",
+								PrimaryKey: true,
+							},
+							{
+								Name:   "name",
+								Type:   "varchar(255)",
+								Unique: true,
+							},
+						},
+					},
+				},
+			},
+			{
+				Name: "02_drop_table",
+				Operations: migrations.Operations{
+					&migrations.OpDropTable{
+						Name: "users",
+					},
+				},
+			},
+		},
+		afterStart: func(t *testing.T, db *sql.DB) {
+			// The view for the deleted table does not exist in the new version schema.
+			ViewMustNotExist(t, db, "public", "02_drop_table", "users")
+
+			// But the underlying table has not been deleted.
+			TableMustExist(t, db, "public", "users")
+		},
+		afterRollback: func(t *testing.T, db *sql.DB) {
+			// Rollback is a no-op.
+		},
+		afterComplete: func(t *testing.T, db *sql.DB) {
+			// The underlying table has been deleted.
+			TableMustNotExist(t, db, "public", "users")
+		},
+	}})
+}

pkg/schema/schema.go

@@ -80,6 +80,10 @@ func (s *Schema) RenameTable(from, to string) error {
 	return nil
 }
 
+func (s *Schema) RemoveTable(name string) {
+	delete(s.Tables, name)
+}
+
 func (t *Table) GetColumn(name string) *Column {
 	if t.Columns == nil {
 		return nil