fix: address audit execution failures and unsafe type conversions

pkg/writers/pgsql/README.md

@@ -346,3 +346,67 @@ metadata:
 - Default values are properly quoted
 - Constraint names follow PostgreSQL conventions
 - Compatible with PostgreSQL 12+
+
+## Live execution failures
+
+Live PostgreSQL output stops after the first failed statement by default and returns
+`*writers.ExecutionFailure` (also available as `*pgsql.ExecutionFailure`; inspect it
+with `errors.As`). `WriterOptions.ContinueOnError` or job
+`options.continue_on_error: true` attempts later statements but still returns an error
+if any fail, preventing dependent jobs from running. A failed type preflight always
+stops live execution, including when continuation is enabled.
+
+PostgreSQL, SQLite, SQL Server, and the PostgreSQL script runner share execution
+reports through `WriterOptions.Metadata["execution_report"]`. Optional
+`Metadata["report_path"]` writes the report as JSON; write and close failures are
+errors. Reports include each operation's status and original number, plus applied,
+failed, skipped, and rolled-back counts. Blank and comment entries count as skipped.
+Unknown commit/rollback outcomes are explicitly marked `unknown` and are never
+reported as confirmed rollback. Connection failures record all planned operations as
+skipped. Raw scripts have different reporting limits; see the sqlexec README.
+
+Generated DDL uses independent statements by default: earlier changes can remain
+committed after failure. Set `WriterOptions.AtomicExecution = true` to use one
+transaction with PostgreSQL, SQLite, or SQL Server generated DDL. For database output
+jobs, use:
+
+```yaml
+options:
+  atomic_execution: true
+```
+
+Atomic execution and continuation are mutually exclusive. Rollback covers database
+transactional effects; sequence advancement and external effects in SQL expressions
+are outside that guarantee. Concurrency locks, migration history, and resumable
+migration management remain separate work (audit A09).
+
+## Column type conversions
+
+Type changes propagate errors and never implicitly rename or replace a column.
+Before schema changes, full-DDL and diff plans evaluate every existing value through
+the conversion expression and target cast. Diff preflights run before dropping
+constraints/indexes or creating sequences; database-wide preflights run before
+changing any schema. Matching types, missing tables/columns, generated columns, and
+identity columns need no conversion check.
+
+Column metadata in JSON/YAML models supports two optional SQL expressions:
+
+```yaml
+metadata:
+  pgsql_using: "split_part(name, ' ', 1)::integer"
+  pgsql_preflight: "name ~ '^[0-9]+ units$'"
+```
+
+`pgsql_using` replaces the default `column::target_type` expression in `USING`.
+`pgsql_preflight` is a row predicate; false or NULL rejects the conversion. These
+fields must contain nonempty strings. Both are trusted executable SQL, like defaults
+and check expressions; they are not values or a sandbox. Preflights evaluate the
+conversion expression, so use expressions without side effects. New types/extensions
+needed by an existing-column conversion must already be available for preflight.
+
+Preflight checks can race concurrent writes and do not prove that locks, defaults,
+or every dependency will permit the ALTER. The ALTER still validates the conversion;
+use atomic execution when preceding operations also need rollback. Generated SQL
+clients must stop on preflight failure; live RelSpec execution enforces this even
+with continuation enabled. Rename/replacement/backfill operations still require a
+separately reviewed migration.

pkg/writers/pgsql/live_execute_test.go

@@ -3,9 +3,11 @@ package pgsql
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
 	"time"
 
@@ -141,15 +143,16 @@ func TestLive_WriteDatabaseEmptyThenIdenticalThenDrifted(t *testing.T) {
 	}
 }
 
-func TestLive_WriteDatabaseFailedStatementContinues(t *testing.T) {
+func TestLive_WriteDatabaseFailedStatementReturnsError(t *testing.T) {
 	connString := liveWriterConn(t)
 	schema, _ := liveWriterSchema(t, connString)
 	reportPath := filepath.Join(t.TempDir(), "report.json")
 
 	db := liveModel(schema, map[string]string{"bad": "no_such_type_xyz"})
 	rep, err := runLiveWrite(t, connString, db, map[string]interface{}{"full_ddl": true, "report_path": reportPath})
-	if err != nil {
-		t.Fatalf("failed statements must not abort the run: %v", err)
+	var failure *ExecutionFailure
+	if !errors.As(err, &failure) {
+		t.Fatalf("expected structured failure, got: %v", err)
 	}
 	if rep.FailedStatements == 0 || len(rep.Errors) != rep.FailedStatements {
 		t.Fatalf("expected recorded failures: %+v", rep)
@@ -230,7 +233,7 @@ func TestLive_ExecuteStatementsSkipsCommentsAndBlank(t *testing.T) {
 		t.Fatal(err)
 	}
 	r := w.executionReport
-	if r.ExecutedStatements != 2 || r.FailedStatements != 0 || r.TotalStatements != 5 {
+	if r.ExecutedStatements != 2 || r.FailedStatements != 0 || r.TotalStatements != 5 || r.SkippedStatements != 3 {
 		t.Errorf("counts: %+v", r)
 	}
 	if len(r.Schemas) != 1 || r.Schemas[0].Name != schema || len(r.Schemas[0].Tables) != 1 || !r.Schemas[0].Tables[0].Created {
@@ -241,3 +244,174 @@ func TestLive_ExecuteStatementsSkipsCommentsAndBlank(t *testing.T) {
 		t.Errorf("table not created: %v", err)
 	}
 }
+
+func TestLive_ExecuteStatementsFailureModes(t *testing.T) {
+	connString := liveWriterConn(t)
+	for _, continueOnError := range []bool{false, true} {
+		t.Run(fmt.Sprintf("continue=%v", continueOnError), func(t *testing.T) {
+			schema, conn := liveWriterSchema(t, connString)
+			reportPath := filepath.Join(t.TempDir(), "report.json")
+			w := NewWriter(&writers.WriterOptions{
+				ContinueOnError: continueOnError,
+				Metadata:        map[string]interface{}{"report_path": reportPath},
+			})
+			w.executionReport = &ExecutionReport{}
+			err := w.executeStatements([]string{
+				"CREATE SCHEMA " + schema,
+				"SELECT 1 / 0",
+				"CREATE TABLE " + schema + ".after_failure (id int)",
+			}, connString)
+			var failure *ExecutionFailure
+			if !errors.As(err, &failure) || failure.Report != w.executionReport {
+				t.Fatalf("expected structured execution failure, got %v", err)
+			}
+			var report ExecutionReport
+			data, err := os.ReadFile(reportPath)
+			if err != nil {
+				t.Fatal(err)
+			}
+			if err := json.Unmarshal(data, &report); err != nil {
+				t.Fatal(err)
+			}
+			executed, skipped := 1, 1
+			if continueOnError {
+				executed, skipped = 2, 0
+			}
+			if report.TotalStatements != 3 || report.ExecutedStatements != executed || report.FailedStatements != 1 || report.SkippedStatements != skipped || report.EndTime == "" {
+				t.Fatalf("incorrect partial application report: %+v", report)
+			}
+			if len(report.Errors) != 1 || report.Errors[0].StatementNumber != 2 {
+				t.Fatalf("incorrect failure details: %+v", report.Errors)
+			}
+			if exists := columnExists(t, conn, schema, "after_failure", "id"); exists != continueOnError {
+				t.Fatalf("later statement executed=%v, continue=%v", exists, continueOnError)
+			}
+		})
+	}
+}
+
+func TestLive_IncompatibleTypeChangePreservesColumnAndData(t *testing.T) {
+	connString := liveWriterConn(t)
+	schema, conn := liveWriterSchema(t, connString)
+	ctx := context.Background()
+	_, err := conn.Exec(ctx, "CREATE SCHEMA "+schema+"; CREATE TABLE "+schema+".accounts (id integer PRIMARY KEY, name text UNIQUE); INSERT INTO "+schema+".accounts VALUES (1, 'not-an-integer')")
+	if err != nil {
+		t.Fatal(err)
+	}
+	var originalIndex uint32
+	if err := conn.QueryRow(ctx, "SELECT to_regclass($1)::oid", schema+".accounts_name_key").Scan(&originalIndex); err != nil {
+		t.Fatal(err)
+	}
+
+	// Exercise both full-DDL and diff generation against populated data.
+	for _, fullDDL := range []bool{false, true} {
+		t.Run(fmt.Sprintf("full_ddl=%v", fullDDL), func(t *testing.T) {
+			model := liveModel(schema, map[string]string{"name": "integer"})
+			unique := models.InitConstraint("accounts_name_key", models.UniqueConstraint)
+			unique.Columns = []string{"name"}
+			model.Schemas[0].Tables[0].Constraints[unique.Name] = unique
+			report, err := runLiveWrite(t, connString, model, map[string]interface{}{"full_ddl": fullDDL})
+			var failure *ExecutionFailure
+			if !errors.As(err, &failure) || report.FailedStatements != 1 {
+				t.Fatalf("expected failed conversion, got report=%+v err=%v", report, err)
+			}
+			var value, typ string
+			if err := conn.QueryRow(ctx, "SELECT name, pg_typeof(name)::text FROM "+schema+".accounts WHERE id=1").Scan(&value, &typ); err != nil {
+				t.Fatal(err)
+			}
+			if value != "not-an-integer" || typ != "text" {
+				t.Fatalf("original column changed: value=%q type=%q", value, typ)
+			}
+			var columns int
+			if err := conn.QueryRow(ctx, "SELECT count(*) FROM information_schema.columns WHERE table_schema=$1 AND table_name='accounts'", schema).Scan(&columns); err != nil || columns != 2 {
+				t.Fatalf("unexpected replacement column: columns=%d err=%v", columns, err)
+			}
+			var index uint32
+			if err := conn.QueryRow(ctx, "SELECT to_regclass($1)::oid", schema+".accounts_name_key").Scan(&index); err != nil || index != originalIndex {
+				t.Fatalf("unique index changed: index=%d original=%d err=%v", index, originalIndex, err)
+			}
+			if _, err := conn.Exec(ctx, "INSERT INTO "+schema+".accounts VALUES (2, 'not-an-integer')"); err == nil {
+				t.Fatal("unique constraint no longer enforced")
+			}
+		})
+	}
+}
+
+func TestLive_AtomicExecutionRollback(t *testing.T) {
+	connString := liveWriterConn(t)
+	schema, conn := liveWriterSchema(t, connString)
+	w := NewWriter(&writers.WriterOptions{AtomicExecution: true})
+	w.executionReport = &ExecutionReport{}
+	err := w.executeStatements([]string{
+		"-- Schema: " + schema,
+		"CREATE SCHEMA " + schema,
+		"CREATE TABLE " + schema + ".created (id int)",
+		"SELECT 1 / 0",
+		"SELECT 1",
+	}, connString)
+	var failure *ExecutionFailure
+	if !errors.As(err, &failure) {
+		t.Fatalf("expected execution failure: %v", err)
+	}
+	report := failure.Report
+	if report.Outcome != "rolled_back" || report.RolledBackStatements != 2 || report.ExecutedStatements != 0 || report.FailedStatements != 1 || report.SkippedStatements != 2 {
+		t.Fatalf("rollback report: %+v", report)
+	}
+	if report.Schemas[0].Tables[0].Created {
+		t.Fatal("rolled-back table must not be reported created")
+	}
+	if columnExists(t, conn, schema, "created", "id") {
+		t.Fatal("table survived transaction rollback")
+	}
+}
+
+func TestLive_TypePreflightPrecedesDropsEvenWithContinuation(t *testing.T) {
+	connString := liveWriterConn(t)
+	schema, conn := liveWriterSchema(t, connString)
+	_, err := conn.Exec(context.Background(), "CREATE SCHEMA "+schema+"; CREATE TABLE "+schema+".accounts (id integer PRIMARY KEY, name text UNIQUE); INSERT INTO "+schema+".accounts VALUES (1, '123'), (2, 'bad')")
+	if err != nil {
+		t.Fatal(err)
+	}
+	// The desired model would drop the unique constraint. Incompatible data must
+	// be caught first, including bad data after the first row.
+	model := liveModel(schema, map[string]string{"name": "integer"})
+	w := NewWriter(&writers.WriterOptions{ContinueOnError: true, Metadata: map[string]interface{}{"connection_string": connString}})
+	err = w.WriteDatabase(model)
+	var failure *ExecutionFailure
+	if !errors.As(err, &failure) || failure.Report.FailedStatements != 1 || failure.Report.ExecutedStatements != 0 {
+		t.Fatalf("preflight must stop mutations: %+v %v", w.executionReport, err)
+	}
+	var exists bool
+	if err := conn.QueryRow(context.Background(), "SELECT to_regclass($1) IS NOT NULL", schema+".accounts_name_key").Scan(&exists); err != nil || !exists {
+		t.Fatalf("constraint was dropped before preflight: %v", err)
+	}
+}
+
+func TestLive_CustomTypeConversionAndPredicate(t *testing.T) {
+	connString := liveWriterConn(t)
+	for _, fullDDL := range []bool{false, true} {
+		t.Run(fmt.Sprintf("full_ddl=%v", fullDDL), func(t *testing.T) {
+			schema, conn := liveWriterSchema(t, connString)
+			_, err := conn.Exec(context.Background(), "CREATE SCHEMA "+schema+"; CREATE TABLE "+schema+".accounts (id integer PRIMARY KEY, name text); INSERT INTO "+schema+".accounts VALUES (1,'42 units'),(2,'7 units')")
+			if err != nil {
+				t.Fatal(err)
+			}
+			model := liveModel(schema, map[string]string{"name": "integer"})
+			col := model.Schemas[0].Tables[0].Columns["name"]
+			col.Metadata["pgsql_using"] = "split_part(name, ' ', 1)::integer"
+			col.Metadata["pgsql_preflight"] = "name ~ '^[0-9]+ units$' AND name NOT LIKE '7 %'"
+			meta := map[string]interface{}{"full_ddl": fullDDL}
+			if _, err := runLiveWrite(t, connString, model, meta); err == nil || !strings.Contains(err.Error(), "preflight rejected rows") {
+				t.Fatalf("predicate must reject: %v", err)
+			}
+			col.Metadata["pgsql_preflight"] = "name ~ '^[0-9]+ units$'"
+			if _, err := runLiveWrite(t, connString, model, meta); err != nil {
+				t.Fatal(err)
+			}
+			var value int
+			if err := conn.QueryRow(context.Background(), "SELECT sum(name) FROM "+schema+".accounts").Scan(&value); err != nil || value != 49 {
+				t.Fatalf("custom conversion result %d: %v", value, err)
+			}
+		})
+	}
+}

pkg/writers/pgsql/migration_writer.go

@@ -487,15 +487,16 @@ func (w *MigrationWriter) generateAlterTableScripts(schema *models.Schema, model
 		} else if !columnsEqual(modelCol, currentCol) {
 			// Column exists but properties changed
 			if !columnTypesEqual(modelCol, currentCol) {
-				newType := effectiveAlterColumnSQLType(modelCol)
-				sql, err := w.executor.ExecuteAlterColumnTypeWithCheck(AlterColumnTypeWithCheckData{
-					SchemaName:      schema.Name,
-					TableName:       modelTable.Name,
-					ColumnName:      modelCol.Name,
-					NewType:         newType,
-					EquivalentTypes: equivalentTypeListSQL(newType),
-					UsingExpr:       buildAlterColumnUsingExpression(modelCol.Name, newType),
+				data, err := conversionData(schema.Name, modelTable.Name, modelCol)
+				if err != nil {
+					return nil, err
+				}
+				scripts = append(scripts, MigrationScript{
+					ObjectName: fmt.Sprintf("%s.%s.%s", schema.Name, modelTable.Name, modelCol.Name),
+					ObjectType: "preflight column type", Schema: schema.Name, Priority: -10000,
+					Sequence: len(scripts), Body: preflightColumnType(data, w.options.FlattenSchema),
 				})
+				sql, err := w.executor.ExecuteAlterColumnTypeWithCheck(data)
 				if err != nil {
 					return nil, err
 				}

pkg/writers/pgsql/migration_writer_test.go

@@ -136,7 +136,7 @@ func TestWriteMigration_AltersColumnTypeWhenActualTypeDiffers(t *testing.T) {
 	}
 }
 
-func TestWriteMigration_AltersColumnTypeFallsBackToRenameAndAddOnConversionFailure(t *testing.T) {
+func TestWriteMigration_AltersColumnTypePreservesConversionFailure(t *testing.T) {
 	current := models.InitDatabase("testdb")
 	currentSchema := models.InitSchema("public")
 	currentTable := models.InitTable("learnings", "public")
@@ -167,17 +167,13 @@ func TestWriteMigration_AltersColumnTypeFallsBackToRenameAndAddOnConversionFailu
 	}
 
 	output := buf.String()
-	if !strings.Contains(output, "EXCEPTION WHEN OTHERS THEN") {
-		t.Fatalf("expected migration to guard the type conversion with an exception handler, got:\n%s", output)
+	if !strings.Contains(output, "ALTER COLUMN details TYPE integer USING details::integer;") {
+		t.Fatalf("expected explicit type conversion, got:\n%s", output)
 	}
-	if !strings.Contains(output, "RENAME COLUMN details TO %I") {
-		t.Fatalf("expected migration to rename the old column (derived from the live type) on conversion failure, got:\n%s", output)
-	}
-	if !strings.Contains(output, "renamed_column := 'details_' || trim(both '_' from regexp_replace(lower(current_type)") {
-		t.Fatalf("expected migration to derive the renamed column name from the live type, got:\n%s", output)
-	}
-	if !strings.Contains(output, "ADD COLUMN details integer") {
-		t.Fatalf("expected migration to add a fresh column with the new type on conversion failure, got:\n%s", output)
+	for _, unsafe := range []string{"EXCEPTION WHEN OTHERS", "renamed_column", "RENAME COLUMN"} {
+		if strings.Contains(output, unsafe) {
+			t.Fatalf("type conversion must propagate errors without replacing the column (%s):\n%s", unsafe, output)
+		}
 	}
 }
 

pkg/writers/pgsql/statement_helpers_test.go

@@ -2,6 +2,7 @@ package pgsql
 
 import (
 	"encoding/json"
+	"errors"
 	"os"
 	"path/filepath"
 	"strings"
@@ -237,8 +238,9 @@ func TestWriteAndFinishReport(t *testing.T) {
 		options:         &writers.WriterOptions{Metadata: map[string]interface{}{"report_path": path}},
 		executionReport: report,
 	}
-	if err := w.finishReport(); err != nil {
-		t.Fatalf("finishReport: %v", err)
+	var failure *ExecutionFailure
+	if err := w.finishReport(); !errors.As(err, &failure) || failure.Report != report {
+		t.Fatalf("expected structured execution failure, got: %v", err)
 	}
 
 	data, err := os.ReadFile(path)
@@ -270,10 +272,15 @@ func TestWriteReportBadPath(t *testing.T) {
 	if err := w.writeReport(filepath.Join(t.TempDir(), "missing", "r.json")); err == nil {
 		t.Error("expected error for unwritable path")
 	}
-	// finishReport must swallow the report error.
+	// Explicitly requested reports are part of successful execution.
 	w.options.Metadata = map[string]interface{}{"report_path": filepath.Join(t.TempDir(), "missing", "r.json")}
-	if err := w.finishReport(); err != nil {
-		t.Errorf("finishReport must not fail on report write error: %v", err)
+	if err := w.finishReport(); err == nil {
+		t.Error("finishReport must fail on report write error")
+	}
+	w.executionReport.FailedStatements = 1
+	var failure *ExecutionFailure
+	if err := w.finishReport(); !errors.As(err, &failure) || !strings.Contains(err.Error(), "requested execution report") {
+		t.Fatalf("must preserve both execution and report errors: %v", err)
 	}
 }
 

pkg/writers/pgsql/templates.go

@@ -99,15 +99,15 @@ type AddColumnData struct {
 
 // AlterColumnTypeWithCheckData contains data for the guarded alter column
 // type template, which only alters existing columns whose live type
-// differs from the desired one, and falls back to renaming the old column
-// and adding a fresh one when the in-place conversion is not possible.
+// differs from the desired one. Conversion errors propagate unchanged.
 type AlterColumnTypeWithCheckData struct {
-	SchemaName      string
-	TableName       string
-	ColumnName      string
-	NewType         string
-	EquivalentTypes string
-	UsingExpr       string
+	SchemaName         string
+	TableName          string
+	ColumnName         string
+	NewType            string
+	EquivalentTypes    string
+	UsingExpr          string
+	PreflightPredicate string
 }
 
 // AlterColumnDefaultWithCheckData contains data for the guarded alter
@@ -332,7 +332,10 @@ func (te *TemplateExecutor) ExecuteAlterColumnTypeWithCheck(data AlterColumnType
 	if err != nil {
 		return "", fmt.Errorf("failed to execute alter_column_type_with_check template: %w", err)
 	}
-	return buf.String(), nil
+	body := strings.TrimSpace(buf.String())
+	body = strings.TrimPrefix(body, "DO $$")
+	body = strings.TrimSuffix(body, "$$;")
+	return dollarBlock(strings.TrimSpace(body), "relspec_alter_type"), nil
 }
 
 // ExecuteAlterColumnDefaultWithCheck executes the guarded alter column

pkg/writers/pgsql/templates/add_column_with_check.tmpl

@@ -1,12 +1 @@
-DO $$
-BEGIN
-  IF NOT EXISTS (
-    SELECT 1 FROM information_schema.columns
-    WHERE table_schema = '{{.SchemaName}}'
-      AND table_name = '{{.TableName}}'
-      AND column_name = '{{.ColumnName}}'
-  ) THEN
-    ALTER TABLE {{qual_table .SchemaName .TableName}} ADD COLUMN {{.ColumnDefinition}};
-  END IF;
-END;
-$$;
+ALTER TABLE {{qual_table .SchemaName .TableName}} ADD COLUMN IF NOT EXISTS {{.ColumnDefinition}};

pkg/writers/pgsql/templates/alter_column_type_with_check.tmpl

@@ -1,30 +1,22 @@
 DO $$
 DECLARE
   current_type text;
-  renamed_column text;
 BEGIN
   SELECT pg_catalog.format_type(a.atttypid, a.atttypmod)
     INTO current_type
   FROM pg_attribute a
   JOIN pg_class t ON t.oid = a.attrelid
   JOIN pg_namespace n ON n.oid = t.relnamespace
-  WHERE n.nspname = '{{.SchemaName}}'
-    AND t.relname = '{{.TableName}}'
-    AND a.attname = '{{.ColumnName}}'
+  WHERE n.nspname = {{quote .SchemaName}}
+    AND t.relname = {{quote .TableName}}
+    AND a.attname = {{quote .ColumnName}}
     AND a.attnum > 0
     AND NOT a.attisdropped;
 
   IF current_type IS NOT NULL
      AND current_type <> ALL(ARRAY[{{.EquivalentTypes}}]) THEN
-    BEGIN
-      ALTER TABLE {{qual_table .SchemaName .TableName}}
-        ALTER COLUMN {{quote_ident .ColumnName}} TYPE {{.NewType}}{{if .UsingExpr}} USING {{.UsingExpr}}{{end}};
-    EXCEPTION WHEN OTHERS THEN
-      renamed_column := '{{.ColumnName}}_' || trim(both '_' from regexp_replace(lower(current_type), '[^a-z0-9]+', '_', 'g'));
-      EXECUTE format('ALTER TABLE {{qual_table .SchemaName .TableName}} RENAME COLUMN {{quote_ident .ColumnName}} TO %I', renamed_column);
-      ALTER TABLE {{qual_table .SchemaName .TableName}}
-        ADD COLUMN {{quote_ident .ColumnName}} {{.NewType}};
-    END;
+    ALTER TABLE {{qual_table .SchemaName .TableName}}
+      ALTER COLUMN {{quote_ident .ColumnName}} TYPE {{.NewType}}{{if .UsingExpr}} USING {{.UsingExpr}}{{end}};
   END IF;
 END;
 $$;

pkg/writers/pgsql/type_conversion.go

@@ -0,0 +1,91 @@
+package pgsql
+
+import (
+	"fmt"
+	"strings"
+
+	"github.com/jackc/pgx/v5"
+
+	"git.warky.dev/wdevs/relspecgo/pkg/models"
+)
+
+// conversionData accepts executable SQL only through explicitly named column metadata.
+func conversionData(schema, table string, col *models.Column) (AlterColumnTypeWithCheckData, error) {
+	target := effectiveAlterColumnSQLType(col)
+	data := AlterColumnTypeWithCheckData{
+		SchemaName: schema, TableName: table, ColumnName: col.Name,
+		NewType: target, EquivalentTypes: equivalentTypeListSQL(target),
+		UsingExpr: buildAlterColumnUsingExpression(col.Name, target),
+	}
+	for key, destination := range map[string]*string{"pgsql_using": &data.UsingExpr, "pgsql_preflight": &data.PreflightPredicate} {
+		if value, exists := col.Metadata[key]; exists {
+			expression, ok := value.(string)
+			if !ok || strings.TrimSpace(expression) == "" {
+				return data, fmt.Errorf("%s.%s.%s metadata %s must be a nonempty SQL expression", schema, table, col.Name, key)
+			}
+			*destination = expression
+		}
+	}
+	return data, nil
+}
+
+// sqlLiteral is independent of the connection's standard_conforming_strings setting.
+func sqlLiteral(value string) string {
+	return "E'" + strings.NewReplacer("\\", "\\\\", "'", "''").Replace(value) + "'"
+}
+
+func dollarBlock(body, prefix string) string {
+	for i := 0; ; i++ {
+		delimiter := fmt.Sprintf("$%s_%d$", prefix, i)
+		if !strings.Contains(body, delimiter) {
+			return "DO " + delimiter + "\n" + body + "\n" + delimiter + ";"
+		}
+	}
+}
+
+// preflightColumnType checks every value before any schema changes. It is advisory:
+// writes can race the check, so the ALTER itself still performs the real conversion.
+func preflightColumnType(data AlterColumnTypeWithCheckData, flatten bool) string {
+	ident := pgx.Identifier{data.TableName}
+	if data.SchemaName != "" {
+		if flatten {
+			ident = pgx.Identifier{data.SchemaName + "_" + data.TableName}
+		} else {
+			ident = pgx.Identifier{data.SchemaName, data.TableName}
+		}
+	}
+	table := ident.Sanitize()
+	body := fmt.Sprintf(`DECLARE current_type text;
+BEGIN
+  SELECT pg_catalog.format_type(atttypid, atttypmod) INTO current_type
+  FROM pg_catalog.pg_attribute
+  WHERE attrelid = to_regclass(%s) AND attname = %s AND attnum > 0 AND NOT attisdropped;
+  IF current_type IS NOT NULL AND current_type <> ALL(ARRAY[%s]) THEN
+`, sqlLiteral(table), sqlLiteral(data.ColumnName), data.EquivalentTypes)
+	if data.PreflightPredicate != "" {
+		body += fmt.Sprintf(`    IF EXISTS (SELECT 1 FROM %s WHERE NOT COALESCE((%s), false)) THEN
+      RAISE EXCEPTION USING MESSAGE = %s;
+    END IF;
+`, table, data.PreflightPredicate, sqlLiteral("type conversion preflight rejected rows for "+data.SchemaName+"."+data.TableName+"."+data.ColumnName))
+	}
+	// Aggregate evaluation forces the conversion for all rows, including non-first rows.
+	body += fmt.Sprintf("    PERFORM count((%s)::%s) FROM %s;\n  END IF;\nEND;", data.UsingExpr, data.NewType, table)
+	return dollarBlock(body, "relspec_preflight")
+}
+
+func (w *Writer) generateTypePreflights(schema *models.Schema) ([]string, error) {
+	var statements []string
+	for _, table := range schema.Tables {
+		for _, col := range getSortedColumns(table.Columns) {
+			if col.Generated || col.Identity {
+				continue
+			}
+			data, err := conversionData(schema.Name, table.Name, col)
+			if err != nil {
+				return nil, err
+			}
+			statements = append(statements, preflightColumnType(data, w.options.FlattenSchema))
+		}
+	}
+	return statements, nil
+}

pkg/writers/pgsql/type_conversion_test.go

@@ -0,0 +1,72 @@
+package pgsql
+
+import (
+	"strings"
+	"testing"
+
+	"git.warky.dev/wdevs/relspecgo/pkg/models"
+	"git.warky.dev/wdevs/relspecgo/pkg/writers"
+)
+
+func TestConversionMetadataValidationAndBlockDelimiters(t *testing.T) {
+	col := models.InitColumn("name", "accounts", "public")
+	col.Type = "integer"
+	for _, key := range []string{"pgsql_using", "pgsql_preflight"} {
+		for _, value := range []interface{}{true, "  "} {
+			col.Metadata[key] = value
+			if _, err := conversionData("public", "accounts", col); err == nil || !strings.Contains(err.Error(), key) {
+				t.Fatalf("expected invalid %s metadata error: %v", key, err)
+			}
+			delete(col.Metadata, key)
+		}
+	}
+	col.Metadata["pgsql_using"] = "length('$relspec_alter_type_0$ $relspec_preflight_0$')"
+	data, err := conversionData("public", "accounts", col)
+	if err != nil {
+		t.Fatal(err)
+	}
+	executor, err := NewTemplateExecutor(false)
+	if err != nil {
+		t.Fatal(err)
+	}
+	alter, err := executor.ExecuteAlterColumnTypeWithCheck(data)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !strings.HasPrefix(alter, "DO $relspec_alter_type_1$") {
+		t.Fatalf("unsafe alter delimiter: %s", alter)
+	}
+	preflight := preflightColumnType(data, false)
+	if !strings.HasPrefix(preflight, "DO $relspec_preflight_1$") {
+		t.Fatalf("unsafe preflight delimiter: %s", preflight)
+	}
+}
+
+func TestPreflightsPrecedeAllSchemaMutations(t *testing.T) {
+	db := models.InitDatabase("preflight")
+	for _, name := range []string{"one", "two"} {
+		schema := models.InitSchema(name)
+		table := models.InitTable("accounts", name)
+		col := models.InitColumn("name", "accounts", name)
+		col.Type = "integer"
+		table.Columns[col.Name] = col
+		schema.Tables = append(schema.Tables, table)
+		db.Schemas = append(db.Schemas, schema)
+	}
+	stmts, err := NewWriter(&writers.WriterOptions{}).GenerateDatabaseStatements(db)
+	if err != nil {
+		t.Fatal(err)
+	}
+	lastPreflight, firstDDL := -1, len(stmts)
+	for i, stmt := range stmts {
+		if strings.HasPrefix(stmt, "DO $relspec_preflight") {
+			lastPreflight = i
+		}
+		if strings.HasPrefix(stmt, "CREATE SCHEMA") && i < firstDDL {
+			firstDDL = i
+		}
+	}
+	if lastPreflight < 0 || lastPreflight >= firstDDL {
+		t.Fatal("preflights must precede all schema changes")
+	}
+}

pkg/writers/pgsql/writer.go

@@ -2,7 +2,6 @@ package pgsql
 
 import (
 	"context"
-	"encoding/json"
 	"fmt"
 	"io"
 	"os"
@@ -12,6 +11,8 @@ import (
 	"sync"
 	"time"
 
+	"github.com/jackc/pgx/v5"
+
 	"git.warky.dev/wdevs/relspecgo/pkg/models"
 	"git.warky.dev/wdevs/relspecgo/pkg/pgsql"
 	"git.warky.dev/wdevs/relspecgo/pkg/readers"
@@ -27,36 +28,12 @@ type Writer struct {
 	executor        *TemplateExecutor
 }
 
-// ExecutionReport tracks the execution status of SQL statements
-type ExecutionReport struct {
-	TotalStatements    int              `json:"total_statements"`
-	ExecutedStatements int              `json:"executed_statements"`
-	FailedStatements   int              `json:"failed_statements"`
-	Schemas            []SchemaReport   `json:"schemas"`
-	Errors             []ExecutionError `json:"errors,omitempty"`
-	StartTime          string           `json:"start_time"`
-	EndTime            string           `json:"end_time"`
-}
-
-// SchemaReport tracks execution per schema
-type SchemaReport struct {
-	Name   string        `json:"name"`
-	Tables []TableReport `json:"tables"`
-}
-
-// TableReport tracks execution per table
-type TableReport struct {
-	Name    string `json:"name"`
-	Created bool   `json:"created"`
-	Error   string `json:"error,omitempty"`
-}
-
-// ExecutionError represents a failed statement
-type ExecutionError struct {
-	StatementNumber int    `json:"statement_number"`
-	Statement       string `json:"statement"`
-	Error           string `json:"error"`
-}
+// Shared execution results retain the PostgreSQL package's public type names.
+type ExecutionReport = writers.ExecutionReport
+type ExecutionFailure = writers.ExecutionFailure
+type SchemaReport = writers.SchemaReport
+type TableReport = writers.TableReport
+type ExecutionError = writers.ExecutionError
 
 // NewWriter creates a new PostgreSQL SQL writer
 func NewWriter(options *writers.WriterOptions) *Writer {
@@ -109,9 +86,20 @@ func (w *Writer) WriteDatabase(db *models.Database) error {
 	}
 	fmt.Fprintf(w.writer, "\n")
 
+	for _, schema := range db.Schemas {
+		preflights, err := w.generateTypePreflights(schema)
+		if err != nil {
+			return err
+		}
+		for _, stmt := range preflights {
+			if _, err := fmt.Fprintln(w.writer, stmt); err != nil {
+				return err
+			}
+		}
+	}
 	// Process each schema in the database
 	for _, schema := range db.Schemas {
-		if err := w.WriteSchema(schema); err != nil {
+		if err := w.writeSchema(schema); err != nil {
 			return fmt.Errorf("failed to write schema %s: %w", schema.Name, err)
 		}
 	}
@@ -129,9 +117,16 @@ func (w *Writer) GenerateDatabaseStatements(db *models.Database) ([]string, erro
 	statements = append(statements, fmt.Sprintf("-- Database: %s", db.Name))
 	statements = append(statements, "-- Generated by RelSpec")
 
+	for _, schema := range db.Schemas {
+		preflights, err := w.generateTypePreflights(schema)
+		if err != nil {
+			return nil, err
+		}
+		statements = append(statements, preflights...)
+	}
 	// Process each schema in the database
 	for _, schema := range db.Schemas {
-		schemaStatements, err := w.GenerateSchemaStatements(schema)
+		schemaStatements, err := w.generateSchemaStatements(schema)
 		if err != nil {
 			return nil, fmt.Errorf("failed to generate statements for schema %s: %w", schema.Name, err)
 		}
@@ -187,6 +182,15 @@ func (w *Writer) primaryKeySetvalStatement(schema *models.Schema, table *models.
 
 // GenerateSchemaStatements generates SQL statements as a list for a single schema
 func (w *Writer) GenerateSchemaStatements(schema *models.Schema) ([]string, error) {
+	preflights, err := w.generateTypePreflights(schema)
+	if err != nil {
+		return nil, err
+	}
+	statements, err := w.generateSchemaStatements(schema)
+	return append(preflights, statements...), err
+}
+
+func (w *Writer) generateSchemaStatements(schema *models.Schema) ([]string, error) {
 	statements := []string{}
 
 	if err := w.checkDirectives(schema); err != nil {
@@ -510,15 +514,11 @@ func (w *Writer) GenerateAlterColumnTypeStatements(schema *models.Schema) ([]str
 				// rather than risk an ALTER COLUMN TYPE Postgres won't accept cleanly.
 				continue
 			}
-			targetType := effectiveAlterColumnSQLType(col)
-			stmt, err := w.executor.ExecuteAlterColumnTypeWithCheck(AlterColumnTypeWithCheckData{
-				SchemaName:      schema.Name,
-				TableName:       table.Name,
-				ColumnName:      col.Name,
-				NewType:         targetType,
-				EquivalentTypes: equivalentTypeListSQL(targetType),
-				UsingExpr:       buildAlterColumnUsingExpression(col.Name, targetType),
-			})
+			data, err := conversionData(schema.Name, table.Name, col)
+			if err != nil {
+				return nil, err
+			}
+			stmt, err := w.executor.ExecuteAlterColumnTypeWithCheck(data)
 			if err != nil {
 				return nil, fmt.Errorf("failed to generate alter column type for %s.%s.%s: %w", schema.Name, table.Name, col.Name, err)
 			}
@@ -774,6 +774,22 @@ func (w *Writer) WriteSchema(schema *models.Schema) error {
 	if w.writer == nil {
 		w.writer = os.Stdout
 	}
+	preflights, err := w.generateTypePreflights(schema)
+	if err != nil {
+		return err
+	}
+	for _, stmt := range preflights {
+		if _, err := fmt.Fprintln(w.writer, stmt); err != nil {
+			return err
+		}
+	}
+	return w.writeSchema(schema)
+}
+
+func (w *Writer) writeSchema(schema *models.Schema) error {
+	if w.writer == nil {
+		w.writer = os.Stdout
+	}
 
 	if err := w.checkDirectives(schema); err != nil {
 		return err
@@ -2054,7 +2070,8 @@ func extractSequenceName(defaultExpr string) string {
 func (w *Writer) executeDatabaseSQL(db *models.Database, connString string) error {
 	// Initialize execution report
 	w.executionReport = &ExecutionReport{
-		StartTime: getCurrentTimestamp(),
+		Dialect:   "PostgreSQL",
+		StartTime: time.Now().Format(time.RFC3339Nano),
 		Schemas:   make([]SchemaReport, 0),
 		Errors:    make([]ExecutionError, 0),
 	}
@@ -2067,7 +2084,8 @@ func (w *Writer) executeDatabaseSQL(db *models.Database, connString string) erro
 		} else {
 			statements = diffStatements
 			if len(statements) == 0 {
-				w.executionReport.EndTime = getCurrentTimestamp()
+				w.executionReport.EndTime = time.Now().Format(time.RFC3339Nano)
+				w.executionReport.Outcome = "no_changes"
 				fmt.Fprintln(os.Stderr, "✓ Database is already up to date; nothing to execute")
 				return w.finishReport()
 			}
@@ -2179,12 +2197,17 @@ func (w *Writer) diffStatements(model, current *models.Database) ([]string, erro
 		return nil, fmt.Errorf("failed to generate migration scripts: %w", err)
 	}
 
+	var preflights []string
 	lastSchema := ""
 	for _, script := range scripts {
 		body := strings.TrimSpace(script.Body)
 		if body == "" {
 			continue
 		}
+		if script.ObjectType == "preflight column type" {
+			preflights = append(preflights, body)
+			continue
+		}
 		if script.Schema != "" && script.Schema != lastSchema {
 			statements = append(statements, fmt.Sprintf("-- Schema: %s", script.Schema))
 			lastSchema = script.Schema
@@ -2192,150 +2215,84 @@ func (w *Writer) diffStatements(model, current *models.Database) ([]string, erro
 		statements = append(statements, body)
 	}
 	statements = append(statements, setvalStatements...)
+	statements = append(preflights, statements...)
 
 	return statements, nil
 }
 
 // executeStatements runs statements one by one against the database and writes the report.
 func (w *Writer) executeStatements(statements []string, connString string) error {
-	w.executionReport.TotalStatements = len(statements)
-
-	// Connect to database
-	fmt.Fprintln(os.Stderr, "      → Connecting to PostgreSQL output database...")
 	ctx := context.Background()
 	conn, err := pgsql.Connect(ctx, connString, "writer-pgsql")
 	if err != nil {
-		return fmt.Errorf("failed to connect to database: %w", err)
+		err = writers.FailExecution(w.options, "PostgreSQL", writers.StatementOperations(statements), fmt.Errorf("failed to connect to database: %w", err))
+		w.executionReport = w.options.Metadata["execution_report"].(*ExecutionReport)
+		return err
 	}
 	defer conn.Close(ctx)
-	fmt.Fprintln(os.Stderr, "      → Connected. Executing statements...")
-
-	// Track schemas and tables
-	schemaMap := make(map[string]*SchemaReport)
+	w.executionReport.Dialect = "PostgreSQL"
+	var tx pgx.Tx
+	exec := func(ctx context.Context, i int, stmt string) error {
+		fmt.Fprintf(os.Stderr, "Executing statement %d/%d [%s] %s...\n", i+1, len(statements), detectStatementType(stmt), extractStatementContext(stmt))
+		if tx != nil {
+			_, err := tx.Exec(ctx, stmt)
+			return err
+		}
+		_, err := conn.Exec(ctx, stmt)
+		return err
+	}
+	backend := writers.ExecutionBackend{
+		Exec:     exec,
+		Begin:    func(ctx context.Context) error { var err error; tx, err = conn.Begin(ctx); return err },
+		Commit:   func(ctx context.Context) error { return tx.Commit(ctx) },
+		Rollback: func(ctx context.Context) error { return tx.Rollback(ctx) },
+	}
+	schemaNames := make([]string, len(statements))
 	currentSchema := ""
-
-	// Execute each statement
 	for i, stmt := range statements {
-		stmtTrimmed := strings.TrimSpace(stmt)
-
-		// Skip comments
-		if strings.HasPrefix(stmtTrimmed, "--") {
-			// Check if this is a schema comment to track schema changes
-			if strings.Contains(stmtTrimmed, "Schema:") {
-				parts := strings.Split(stmtTrimmed, "Schema:")
-				if len(parts) > 1 {
-					currentSchema = strings.TrimSpace(parts[1])
-					if _, exists := schemaMap[currentSchema]; !exists {
-						schemaReport := SchemaReport{
-							Name:   currentSchema,
-							Tables: make([]TableReport, 0),
-						}
-						schemaMap[currentSchema] = &schemaReport
-					}
-				}
-			}
-			continue
-		}
-
-		// Skip empty statements
-		if stmtTrimmed == "" {
-			continue
-		}
-
-		stmtType := detectStatementType(stmtTrimmed)
-		stmtCtx := extractStatementContext(stmtTrimmed)
-		if stmtCtx != "" {
-			fmt.Fprintf(os.Stderr, "Executing statement %d/%d [%s] %s...\n", i+1, len(statements), stmtType, stmtCtx)
-		} else {
-			fmt.Fprintf(os.Stderr, "Executing statement %d/%d [%s]...\n", i+1, len(statements), stmtType)
-		}
-
-		_, execErr := conn.Exec(ctx, stmt)
-		if execErr != nil {
-			w.executionReport.FailedStatements++
-			execError := ExecutionError{
-				StatementNumber: i + 1,
-				Statement:       truncateStatement(stmt),
-				Error:           execErr.Error(),
-			}
-			w.executionReport.Errors = append(w.executionReport.Errors, execError)
-
-			// Track table creation failure
-			if strings.Contains(strings.ToUpper(stmtTrimmed), "CREATE TABLE") && currentSchema != "" {
-				tableName := extractTableNameFromCreate(stmtTrimmed)
-				if tableName != "" && schemaMap[currentSchema] != nil {
-					schemaMap[currentSchema].Tables = append(schemaMap[currentSchema].Tables, TableReport{
-						Name:    tableName,
-						Created: false,
-						Error:   execErr.Error(),
-					})
-				}
-			}
-
-			// Continue with next statement instead of failing completely
-			fmt.Fprintf(os.Stderr, "⚠ Warning: Statement %d failed: %v\n", i+1, execErr)
-			continue
-		}
-
-		w.executionReport.ExecutedStatements++
-
-		// Track successful table creation
-		if strings.Contains(strings.ToUpper(stmtTrimmed), "CREATE TABLE") && currentSchema != "" {
-			tableName := extractTableNameFromCreate(stmtTrimmed)
-			if tableName != "" && schemaMap[currentSchema] != nil {
-				schemaMap[currentSchema].Tables = append(schemaMap[currentSchema].Tables, TableReport{
-					Name:    tableName,
-					Created: true,
-				})
-			}
+		if strings.HasPrefix(strings.TrimSpace(stmt), "-- Schema:") {
+			currentSchema = strings.TrimSpace(strings.TrimPrefix(strings.TrimSpace(stmt), "-- Schema:"))
 		}
+		schemaNames[i] = currentSchema
 	}
-
-	// Convert schema map to slice
-	for _, schemaReport := range schemaMap {
-		w.executionReport.Schemas = append(w.executionReport.Schemas, *schemaReport)
+	schemaIndexes := make(map[string]int)
+	// Populate per-schema details before finalizing the shared report.
+	backend.Exec = func(ctx context.Context, i int, stmt string) error {
+		err := exec(ctx, i, stmt)
+		if strings.Contains(strings.ToUpper(stmt), "CREATE TABLE") {
+			schemaName := schemaNames[i]
+			if schemaName != "" {
+				idx, exists := schemaIndexes[schemaName]
+				if !exists {
+					idx = -1
+				}
+				if idx < 0 {
+					w.executionReport.Schemas = append(w.executionReport.Schemas, SchemaReport{Name: schemaName, Tables: []TableReport{}})
+					idx = len(w.executionReport.Schemas) - 1
+					schemaIndexes[schemaName] = idx
+				}
+				table := TableReport{Name: extractTableNameFromCreate(stmt), Created: err == nil}
+				if err != nil {
+					table.Error = err.Error()
+				}
+				w.executionReport.Schemas[idx].Tables = append(w.executionReport.Schemas[idx].Tables, table)
+			}
+		}
+		return err
 	}
-
-	w.executionReport.EndTime = getCurrentTimestamp()
-	return w.finishReport()
+	return writers.RunExecution(ctx, w.options, w.executionReport, writers.StatementOperations(statements), backend)
 }
 
-// finishReport writes the optional report file and prints the execution summary.
 func (w *Writer) finishReport() error {
-	// Write report if path is specified
-	if reportPath, ok := w.options.Metadata["report_path"].(string); ok && reportPath != "" {
-		if err := w.writeReport(reportPath); err != nil {
-			fmt.Fprintf(os.Stderr, "⚠ Warning: Failed to write report: %v\n", err)
-		} else {
-			fmt.Fprintf(os.Stderr, "✓ Report written to: %s\n", reportPath)
-		}
-	}
-
+	var err error
 	if w.executionReport.FailedStatements > 0 {
-		fmt.Fprintf(os.Stderr, "⚠ Completed with %d errors out of %d statements\n",
-			w.executionReport.FailedStatements, w.executionReport.TotalStatements)
-	} else {
-		fmt.Fprintf(os.Stderr, "✓ Successfully executed %d statements\n", w.executionReport.ExecutedStatements)
+		err = &ExecutionFailure{Report: w.executionReport}
 	}
-
-	return nil
+	return writers.FinishExecution(w.options, w.executionReport, err)
 }
 
-// writeReport writes the execution report to a JSON file
 func (w *Writer) writeReport(reportPath string) error {
-	file, err := os.Create(reportPath)
-	if err != nil {
-		return fmt.Errorf("failed to create report file: %w", err)
-	}
-	defer file.Close()
-
-	encoder := json.NewEncoder(file)
-	encoder.SetIndent("", "  ")
-	if err := encoder.Encode(w.executionReport); err != nil {
-		return fmt.Errorf("failed to encode report: %w", err)
-	}
-
-	return nil
+	return writers.WriteExecutionReport(reportPath, w.executionReport)
 }
 
 // extractTableNameFromCreate extracts table name from CREATE TABLE statement

pkg/writers/pgsql/writer_test.go

@@ -1016,15 +1016,9 @@ func TestGenerateAddColumnStatements(t *testing.T) {
 
 	// Verify expected elements
 	expectedStrings := []string{
-		"ALTER TABLE public.users ADD COLUMN id integer NOT NULL",
-		"ALTER TABLE public.users ADD COLUMN email varchar(255) NOT NULL",
-		"ALTER TABLE public.users ADD COLUMN status text DEFAULT 'active'",
-		"information_schema.columns",
-		"table_schema = 'public'",
-		"table_name = 'users'",
-		"column_name = 'id'",
-		"column_name = 'email'",
-		"column_name = 'status'",
+		"ALTER TABLE public.users ADD COLUMN IF NOT EXISTS id integer NOT NULL",
+		"ALTER TABLE public.users ADD COLUMN IF NOT EXISTS email varchar(255) NOT NULL",
+		"ALTER TABLE public.users ADD COLUMN IF NOT EXISTS status text DEFAULT 'active'",
 	}
 
 	for _, expected := range expectedStrings {
@@ -1033,12 +1027,6 @@ func TestGenerateAddColumnStatements(t *testing.T) {
 		}
 	}
 
-	// Verify DO blocks are present for conditional adds
-	doBlockCount := strings.Count(output, "DO $$")
-	if doBlockCount < 3 {
-		t.Errorf("Expected at least 3 DO blocks (one per column), got %d", doBlockCount)
-	}
-
 	// Verify IF NOT EXISTS logic
 	ifNotExistsCount := strings.Count(output, "IF NOT EXISTS")
 	if ifNotExistsCount < 3 {
@@ -1083,14 +1071,14 @@ func TestWriteAddColumnStatements(t *testing.T) {
 	t.Logf("Generated output:\n%s", output)
 
 	// Verify output contains expected elements
-	if !strings.Contains(output, "ALTER TABLE public.products ADD COLUMN id integer") {
+	if !strings.Contains(output, "ALTER TABLE public.products ADD COLUMN IF NOT EXISTS id integer") {
 		t.Errorf("Output missing ADD COLUMN for id\nFull output:\n%s", output)
 	}
-	if !strings.Contains(output, "ALTER TABLE public.products ADD COLUMN description text") {
+	if !strings.Contains(output, "ALTER TABLE public.products ADD COLUMN IF NOT EXISTS description text") {
 		t.Errorf("Output missing ADD COLUMN for description\nFull output:\n%s", output)
 	}
-	if !strings.Contains(output, "DO $$") {
-		t.Errorf("Output missing DO block\nFull output:\n%s", output)
+	if strings.Contains(output, "information_schema.columns") {
+		t.Errorf("column additions should use native IF NOT EXISTS\nFull output:\n%s", output)
 	}
 }
 
@@ -1206,7 +1194,7 @@ func TestWriteSchema_AlterColumnDefaultStripsBackticksFromFunctionExpression(t *
 	}
 }
 
-func TestWriteSchema_GuardedAlterColumnTypeFallsBackOnConversionFailure(t *testing.T) {
+func TestWriteSchema_GuardedAlterColumnTypePreservesConversionFailure(t *testing.T) {
 	db := models.InitDatabase("testdb")
 	schema := models.InitSchema("public")
 
@@ -1228,17 +1216,13 @@ func TestWriteSchema_GuardedAlterColumnTypeFallsBackOnConversionFailure(t *testi
 	}
 
 	output := buf.String()
-	if !strings.Contains(output, "EXCEPTION WHEN OTHERS THEN") {
-		t.Fatalf("expected guarded alter to fall back on conversion failure, got:\n%s", output)
+	if !strings.Contains(output, "ALTER COLUMN name TYPE integer USING name::integer;") {
+		t.Fatalf("expected explicit type conversion, got:\n%s", output)
 	}
-	if !strings.Contains(output, "renamed_column := 'name_' || trim(both '_' from regexp_replace(lower(current_type)") {
-		t.Fatalf("expected fallback to derive a renamed column name from the live type, got:\n%s", output)
-	}
-	if !strings.Contains(output, "RENAME COLUMN name TO %I") {
-		t.Fatalf("expected fallback to rename the existing column, got:\n%s", output)
-	}
-	if !strings.Contains(output, "ADD COLUMN name integer") {
-		t.Fatalf("expected fallback to add a fresh column with the new type, got:\n%s", output)
+	for _, unsafe := range []string{"EXCEPTION WHEN OTHERS", "renamed_column", "RENAME COLUMN"} {
+		if strings.Contains(output, unsafe) {
+			t.Fatalf("type conversion must propagate errors without replacing the column (%s):\n%s", unsafe, output)
+		}
 	}
 }