Update schema during create table validation (#455)

Update the schema during `create_table` operation validation so that
validation of subsequent operations in the same migration can see the
new table.

This means that migrations that add a new table and then perform some
other operation, like adding a column, on that table can be validated
because the new table is visible to the `add_column` operation during
it's validation phase.

This means that the following migration is now able to validate:

```json
{
  "name": "43_multiple_ops",
  "operations": [
    {
      "create_table": {
        "name": "players",
        "columns": [
          {
            "name": "id",
            "type": "serial",
            "pk": true
          },
          {
            "name": "name",
            "type": "varchar(255)",
            "check": {
              "name": "name_length_check",
              "constraint": "length(name) > 2"
            }
          }
        ]
      }
    },
    {
      "add_column": {
        "table": "players",
        "column": {
          "name": "rating",
          "type": "integer",
          "comment": "hello world",
          "check": {
            "name": "rating_check",
            "constraint": "rating > 0 AND rating < 100"
          },
          "nullable": false
        }
      }
    }
  ]
}
```

Previously, the new table would not have been visible to the
`add_column` operation and its validation would have failed.

In order to make this work the schema needs to be re-read from the
target database in between validation and migration start, as the
previous assumption that validation would not make changes to the
in-memory schema representation no longer holds.

Part of #239

pkg/migrations/op_add_column_test.go

@@ -1524,6 +1524,7 @@ func TestAddColumnToATableCreatedInTheSameMigration(t *testing.T) {
 								Name:     "age",
 								Type:     "integer",
 								Nullable: ptr(false),
+								Default:  ptr("18"),
 								Check: &migrations.CheckConstraint{
 									Name:       "age_check",
 									Constraint: "age >= 18",
@@ -1559,7 +1560,7 @@ func TestAddColumnToATableCreatedInTheSameMigration(t *testing.T) {
 				}, testutils.CheckViolationErrorCode)
 			},
 		},
-	}, roll.WithSkipValidation(true)) // TODO: remove once this migration can be validated
+	})
 }
 
 func TestAddColumnInvalidNameLength(t *testing.T) {

pkg/migrations/op_create_table.go

@@ -47,17 +47,8 @@ func (o *OpCreateTable) Start(ctx context.Context, conn db.DB, latestSchema stri
 		}
 	}
 
-	columns := make(map[string]schema.Column, len(o.Columns))
-	for _, col := range o.Columns {
-		columns[col.Name] = schema.Column{
-			Name: col.Name,
-		}
-	}
-
-	s.AddTable(o.Name, schema.Table{
-		Name:    tempName,
-		Columns: columns,
-	})
+	// Update the in-memory schema representation with the new table
+	o.updateSchema(s)
 
 	return nil, nil
 }
@@ -117,9 +108,30 @@ func (o *OpCreateTable) Validate(ctx context.Context, s *schema.Schema) error {
 		}
 	}
 
+	// Update the schema to ensure that the new table is visible to validation of
+	// subsequent operations.
+	o.updateSchema(s)
+
 	return nil
 }
 
+// updateSchema updates the in-memory schema representation with the details of
+// the new table.
+func (o *OpCreateTable) updateSchema(s *schema.Schema) *schema.Schema {
+	columns := make(map[string]schema.Column, len(o.Columns))
+	for _, col := range o.Columns {
+		columns[col.Name] = schema.Column{
+			Name: col.Name,
+		}
+	}
+	s.AddTable(o.Name, schema.Table{
+		Name:    TemporaryName(o.Name),
+		Columns: columns,
+	})
+
+	return s
+}
+
 func columnsToSQL(cols []Column, tr SQLTransformer) (string, error) {
 	var sql string
 	var primaryKeys []string

pkg/roll/execute.go

@@ -75,6 +75,13 @@ func (m *Roll) StartDDLOperations(ctx context.Context, migration *migrations.Mig
 	}
 	latestSchema := VersionedSchemaName(m.schema, *latestVersion)
 
+	// Reread the latest schema as validation may have updated the schema object
+	// in memory.
+	newSchema, err = m.state.ReadSchema(ctx, m.schema)
+	if err != nil {
+		return nil, fmt.Errorf("unable to read schema: %w", err)
+	}
+
 	// execute operations
 	var tablesToBackfill []*schema.Table
 	for _, op := range migration.Operations {