fix: validate extension schema consistency between plan and target databases (#518)

cmd/plan/external_db_integration_test.go

@@ -103,7 +103,7 @@ func TestExternalDatabase_VersionMismatch(t *testing.T) {
 	targetHost, targetPort, targetDatabase, targetUser, targetPassword := targetDB.GetConnectionDetails()
 
 	// Detect version from target database
-	pgVersion, err := postgres.DetectPostgresVersionFromDB(
+	pgVersion, _, err := postgres.DetectPostgresVersionAndExtensionsFromDB(
 		targetHost,
 		targetPort,
 		targetDatabase,
@@ -115,6 +115,69 @@ func TestExternalDatabase_VersionMismatch(t *testing.T) {
 	assert.NotEmpty(t, pgVersion, "version should not be empty")
 }
 
+// TestExternalDatabase_ExtensionSchemaMismatch tests that provider creation fails fast
+// when an extension is installed in different schemas on the plan and target databases.
+// Extension-owned types (e.g. pgvector's vector, citext) are schema-qualified by whatever
+// schema their extension resolves to, so a mismatch silently produces a wrong plan:
+// spurious ALTER COLUMN TYPE or CREATE TABLE DDL referencing a schema that doesn't
+// exist on the target (issue #518).
+func TestExternalDatabase_ExtensionSchemaMismatch(t *testing.T) {
+	// Skip in short mode
+	if testing.Short() {
+		t.Skip("skipping integration test")
+	}
+
+	targetDB := testutil.SetupPostgres(t)
+	defer targetDB.Stop()
+
+	externalPlanDB := testutil.SetupPostgres(t)
+	defer externalPlanDB.Stop()
+
+	// Install citext into schema "exts" on the target, but "public" on the plan database
+	targetConn, targetHost, targetPort, targetDatabase, targetUser, targetPassword := testutil.ConnectToPostgres(t, targetDB)
+	defer targetConn.Close()
+	_, err := targetConn.Exec("CREATE SCHEMA exts")
+	require.NoError(t, err)
+	_, err = targetConn.Exec("CREATE EXTENSION citext SCHEMA exts")
+	require.NoError(t, err)
+
+	planConn, planHost, planPort, planDatabase, planUser, planPassword := testutil.ConnectToPostgres(t, externalPlanDB)
+	defer planConn.Close()
+	_, err = planConn.Exec("CREATE EXTENSION citext SCHEMA public")
+	require.NoError(t, err)
+
+	config := &PlanConfig{
+		Host:            targetHost,
+		Port:            targetPort,
+		DB:              targetDatabase,
+		User:            targetUser,
+		Password:        targetPassword,
+		Schema:          "public",
+		ApplicationName: "pgschema-test",
+		PlanDBHost:      planHost,
+		PlanDBPort:      planPort,
+		PlanDBDatabase:  planDatabase,
+		PlanDBUser:      planUser,
+		PlanDBPassword:  planPassword,
+	}
+
+	_, err = CreateDesiredStateProvider(config)
+	require.Error(t, err, "should fail when extension schemas differ between plan and target databases")
+	assert.Contains(t, err.Error(), "citext", "error should name the mismatched extension")
+	assert.Contains(t, err.Error(), "exts", "error should name the target schema")
+	assert.Contains(t, err.Error(), "public", "error should name the plan database schema")
+
+	// After moving the extension to the matching schema, provider creation succeeds
+	_, err = planConn.Exec("CREATE SCHEMA exts")
+	require.NoError(t, err)
+	_, err = planConn.Exec("ALTER EXTENSION citext SET SCHEMA exts")
+	require.NoError(t, err)
+
+	provider2, err := CreateDesiredStateProvider(config)
+	require.NoError(t, err, "matching extension schemas should pass validation")
+	provider2.Stop()
+}
+
 // TestExternalDatabase_CleanupOnError tests that temporary schema is cleaned up on errors
 func TestExternalDatabase_CleanupOnError(t *testing.T) {
 	// Skip in short mode

cmd/plan/plan.go

@@ -199,7 +199,9 @@ type PlanConfig struct {
 // for validating the desired state schema. The caller is responsible for calling Stop() on the returned provider.
 func CreateDesiredStateProvider(config *PlanConfig) (postgres.DesiredStateProvider, error) {
 	// Detect target database PostgreSQL version (needed for both embedded and external)
-	pgVersion, err := postgres.DetectPostgresVersionFromDB(
+	// and its extension installation schemas (needed for the external database's
+	// extension schema consistency check, issue #518) in a single connection.
+	pgVersion, targetExtensions, err := postgres.DetectPostgresVersionAndExtensionsFromDB(
 		config.Host,
 		config.Port,
 		config.DB,
@@ -229,6 +231,7 @@ func CreateDesiredStateProvider(config *PlanConfig) (postgres.DesiredStateProvid
 			Password:           config.PlanDBPassword,
 			SSLMode:            config.PlanDBSSLMode,
 			TargetMajorVersion: targetMajorVersion,
+			TargetExtensions:   targetExtensions,
 		}
 		return postgres.NewExternalDatabase(externalConfig)
 	}

internal/postgres/embedded.go

@@ -48,9 +48,11 @@ type EmbeddedPostgresConfig struct {
 	Password string
 }
 
-// DetectPostgresVersionFromDB connects to a database and detects its version
-// This is a convenience function that opens a connection, detects the version, and closes it
-func DetectPostgresVersionFromDB(host string, port int, database, user, password, sslmode string) (PostgresVersion, error) {
+// DetectPostgresVersionAndExtensionsFromDB connects to a database and detects its
+// version along with the installation schema of every installed extension (keyed by
+// extension name). Gathering both over a single connection avoids a second
+// connect/close round trip during plan generation.
+func DetectPostgresVersionAndExtensionsFromDB(host string, port int, database, user, password, sslmode string) (PostgresVersion, map[string]string, error) {
 	// Build connection config
 	finalSSLMode := sslmode
 	if finalSSLMode == "" {
@@ -68,12 +70,22 @@ func DetectPostgresVersionFromDB(host string, port int, database, user, password
 	// Connect to database
 	db, err := util.Connect(config)
 	if err != nil {
-		return "", fmt.Errorf("failed to connect to database: %w", err)
+		return "", nil, fmt.Errorf("failed to connect to database: %w", err)
 	}
 	defer db.Close()
 
 	// Detect version
-	return detectPostgresVersion(db)
+	version, err := detectPostgresVersion(db)
+	if err != nil {
+		return "", nil, err
+	}
+
+	extensions, err := getExtensionSchemas(db)
+	if err != nil {
+		return "", nil, fmt.Errorf("failed to query extensions: %w", err)
+	}
+
+	return version, extensions, nil
 }
 
 // StartEmbeddedPostgres starts a temporary embedded PostgreSQL instance

internal/postgres/external.go

@@ -5,6 +5,8 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"sort"
+	"strings"
 
 	_ "github.com/jackc/pgx/v5/stdlib"
 	"github.com/pgplex/pgschema/cmd/util"
@@ -32,6 +34,14 @@ type ExternalDatabaseConfig struct {
 	Password           string
 	SSLMode            string
 	TargetMajorVersion int // Expected major version to match
+	// TargetExtensions maps extension name to its installation schema on the
+	// target database. Optional: when populated (typically from querying the
+	// target database), NewExternalDatabase validates that extensions installed
+	// on both databases live in the same schema on each side; when empty or nil,
+	// the validation is skipped. Without it, extension-owned types (e.g.
+	// pgvector's vector) installed in different schemas resolve to different
+	// schema-qualified names and the generated plan is silently wrong (issue #518).
+	TargetExtensions map[string]string
 }
 
 // sslModeOrDefault returns the configured SSL mode, defaulting to "prefer" if empty
@@ -77,6 +87,12 @@ func NewExternalDatabase(config *ExternalDatabaseConfig) (*ExternalDatabase, err
 		)
 	}
 
+	// Validate that extensions installed on both databases live in the same schema
+	if err := validateExtensionSchemas(db, config.TargetExtensions); err != nil {
+		db.Close()
+		return nil, err
+	}
+
 	// Generate temporary schema name with unique timestamp
 	tempSchema := GenerateTempSchemaName()
 
@@ -185,6 +201,64 @@ func (ed *ExternalDatabase) Stop() error {
 	return nil
 }
 
+// validateExtensionSchemas ensures every extension installed on both the plan and
+// target databases lives in the same schema on each side. Extension-owned types
+// (e.g. pgvector's vector) are schema-qualified by whatever schema their extension
+// is installed in, so a mismatch makes the two sides' type names differ and yields
+// a silently wrong plan: spurious ALTER COLUMN TYPE, or CREATE TABLE DDL that
+// references a schema that doesn't exist on the target (issue #518).
+// Extensions present on only one side are not an error: if the desired state
+// actually needs a missing extension, applying it to the plan database fails
+// loudly with the existing hint.
+func validateExtensionSchemas(db *sql.DB, targetExtensions map[string]string) error {
+	if len(targetExtensions) == 0 {
+		return nil
+	}
+
+	planExtensions, err := getExtensionSchemas(db)
+	if err != nil {
+		return fmt.Errorf("failed to query plan database extensions: %w", err)
+	}
+
+	var mismatches []string
+	for name, planSchema := range planExtensions {
+		if targetSchema, ok := targetExtensions[name]; ok && targetSchema != planSchema {
+			mismatches = append(mismatches, fmt.Sprintf(
+				"extension %q is installed in schema %q on the plan database but schema %q on the target database",
+				name, planSchema, targetSchema))
+		}
+	}
+	if len(mismatches) == 0 {
+		return nil
+	}
+
+	sort.Strings(mismatches)
+	return fmt.Errorf(
+		"extension schema mismatch: %s; the plan database must mirror the target's extension schemas (use ALTER EXTENSION ... SET SCHEMA or reinstall with CREATE EXTENSION ... WITH SCHEMA), see https://www.pgschema.com/cli/plan-db",
+		strings.Join(mismatches, "; "))
+}
+
+// getExtensionSchemas returns the installation schema of every extension in the
+// connected database, keyed by extension name.
+func getExtensionSchemas(db *sql.DB) (map[string]string, error) {
+	rows, err := db.QueryContext(context.Background(),
+		"SELECT e.extname, n.nspname FROM pg_catalog.pg_extension e JOIN pg_catalog.pg_namespace n ON n.oid = e.extnamespace")
+	if err != nil {
+		return nil, err
+	}
+	defer rows.Close()
+
+	extensions := make(map[string]string)
+	for rows.Next() {
+		var name, schema string
+		if err := rows.Scan(&name, &schema); err != nil {
+			return nil, err
+		}
+		extensions[name] = schema
+	}
+	return extensions, rows.Err()
+}
+
 // detectMajorVersion queries the database to determine its PostgreSQL major version
 func detectMajorVersion(db *sql.DB) (int, error) {
 	ctx := context.Background()