From 9b1285c37d0e95088761ffed1990e225b3f8c20c Mon Sep 17 00:00:00 2001 From: Christoph Ostertag <37454333+christophostertag@users.noreply.github.com> Date: Sat, 19 Sep 2026 18:30:43 +0200 Subject: [PATCH] fix: recover interrupted concurrent index builds (#598) --- cmd/apply/apply.go | 57 +++- cmd/issue_598_integration_test.go | 370 +++++++++++++++++++++++ cmd/plan/plan.go | 8 +- docs/workflow/online-ddl.mdx | 40 +++ internal/diff/index_recovery.go | 107 +++++++ internal/diff/index_recovery_test.go | 78 +++++ internal/fingerprint/index_state_test.go | 35 +++ internal/plan/index_recovery.go | 66 ++++ internal/plan/index_recovery_test.go | 23 ++ internal/plan/rewrite.go | 4 + ir/index_state.go | 61 ++++ ir/inspector.go | 4 + ir/ir.go | 5 +- ir/queries/queries.sql | 41 +++ ir/queries/queries.sql.go | 89 ++++++ 15 files changed, 982 insertions(+), 6 deletions(-) create mode 100644 cmd/issue_598_integration_test.go create mode 100644 internal/diff/index_recovery.go create mode 100644 internal/diff/index_recovery_test.go create mode 100644 internal/fingerprint/index_state_test.go create mode 100644 internal/plan/index_recovery.go create mode 100644 internal/plan/index_recovery_test.go create mode 100644 ir/index_state.go diff --git a/cmd/apply/apply.go b/cmd/apply/apply.go index 28bfda0be..ace3b807f 100644 --- a/cmd/apply/apply.go +++ b/cmd/apply/apply.go @@ -182,15 +182,13 @@ func ApplyMigration(config *ApplyConfig, provider postgres.DesiredStateProvider) return fmt.Errorf("failed to load .pgschemaignore: %w", err) } - // Validate schema fingerprint if plan has one + // Reject a stale saved plan before presenting it for approval. if migrationPlan.SourceFingerprint != nil { - err := validateSchemaFingerprint(migrationPlan, config.Host, config.Port, config.DB, config.User, config.Password, config.SSLMode, config.Schema, config.ApplicationName, ignoreConfig) - if err != nil { + if err := validateSchemaFingerprint(migrationPlan, config.Host, config.Port, config.DB, config.User, config.Password, config.SSLMode, config.Schema, config.ApplicationName, ignoreConfig); err != nil { return err } } - // Check if there are any changes to apply by examining the plan diffs if !migrationPlan.HasAnyChanges() { fmt.Println("No changes to apply. Database schema is already up to date.") return nil @@ -217,6 +215,15 @@ func ApplyMigration(config *ApplyConfig, provider postgres.DesiredStateProvider) } } + // An interactive approval may have waited arbitrarily long. Automatic + // approval already has the fresh check above and needs no duplicate scan. + if !config.AutoApprove && migrationPlan.SourceFingerprint != nil { + err := validateSchemaFingerprint(migrationPlan, config.Host, config.Port, config.DB, config.User, config.Password, config.SSLMode, config.Schema, config.ApplicationName, ignoreConfig) + if err != nil { + return err + } + } + // Apply the changes if !config.Quiet { fmt.Println("\nApplying changes...") @@ -292,6 +299,11 @@ func ApplyMigration(config *ApplyConfig, provider postgres.DesiredStateProvider) err = executeGroup(ctx, conn, group, i+1, config.Quiet, retry) if err != nil { + for _, step := range group.Steps { + if strings.HasPrefix(strings.TrimSpace(step.SQL), "REINDEX INDEX CONCURRENTLY ") { + return indexRecoveryError(err) + } + } return err } } @@ -466,6 +478,27 @@ func validateSchemaFingerprint(migrationPlan *plan.Plan, host string, port int, return fmt.Errorf("failed to get current database state for fingerprint validation: %w", err) } + if currentSchema := currentStateIR.Schemas[schema]; currentSchema != nil { + for name, state := range currentSchema.IndexStates { + indexPath := schema + "." + state.Table + "." + name + tablePath := schema + "." + state.Table + for _, group := range migrationPlan.Groups { + for _, step := range group.Steps { + // Ignore comments and unrelated changes, particularly on a + // legitimate invalid ON ONLY partition parent. + affectsIndex := (step.Type == "table.index" || step.Type == "materialized_view.index") && step.Path == indexPath + affectsConstraint := state.Constraint != "" && step.Type == "table.constraint" && step.Path == tablePath+"."+state.Constraint + removesParent := (step.Operation == "drop" || step.Operation == "recreate") && step.Path == tablePath + if affectsIndex || affectsConstraint || removesParent { + if err := state.CheckOperation(schema, name); err != nil { + return err + } + } + } + } + } + } + // Compute current fingerprint currentFingerprint, err := fingerprint.ComputeFingerprint(currentStateIR, schema) if err != nil { @@ -641,3 +674,19 @@ func truncateSQL(sql string, maxLen int) string { return cleaned[:maxLen-3] + "..." } + +// Recovery must not retry a partly committed concurrent operation in place. +// Keep PostgreSQL's error and direct the caller to a fresh state-aware plan. +func indexRecoveryError(err error) error { + advice := "inspect the interrupted operation and regenerate the plan before retrying" + var pgErr *pgconn.PgError + if errors.As(err, &pgErr) { + switch pgErr.Code { + case "23505": + advice = "resolve the duplicate data explicitly, then regenerate the plan; recovery has not removed the existing index or changed table rows" + case "42501": + advice = "run recovery as the table owner or a role with the required PostgreSQL maintenance privileges, then regenerate the plan" + } + } + return fmt.Errorf("concurrent index recovery failed: %w; %s", err, advice) +} diff --git a/cmd/issue_598_integration_test.go b/cmd/issue_598_integration_test.go new file mode 100644 index 000000000..efc99d078 --- /dev/null +++ b/cmd/issue_598_integration_test.go @@ -0,0 +1,370 @@ +package cmd + +import ( + "context" + "database/sql" + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "testing" + "time" + + "github.com/pgplex/pgschema/cmd/apply" + planCmd "github.com/pgplex/pgschema/cmd/plan" + "github.com/pgplex/pgschema/internal/plan" + "github.com/pgplex/pgschema/internal/postgres" + "github.com/pgplex/pgschema/testutil" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +const issue598Tables = ` +CREATE TABLE parents (id integer PRIMARY KEY); +CREATE TABLE items ( + id integer PRIMARY KEY, + code integer, + payload text NOT NULL UNIQUE, + parent_id integer REFERENCES parents(id) +);` + +// TestIssue598InterruptedConcurrentIndex exercises an actual interrupted native +// plan/apply, rather than manufacturing an invalid catalog entry or repairing it +// with SQL outside pgschema. The writer transaction makes interruption repeatable +// even on small tables and fast machines. +func TestIssue598InterruptedConcurrentIndex(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + provider := testutil.SetupPostgres(t) + defer provider.Stop() + + for _, unique := range []bool{false, true} { + name, modifier := "ordinary", "" + if unique { + name, modifier = "unique", "UNIQUE " + } + t.Run(name, func(t *testing.T) { + indexSQL := fmt.Sprintf("CREATE %sINDEX items_code_idx ON items (code) INCLUDE (payload) WITH (fillfactor=80) WHERE code > 0;", modifier) + if unique { + indexSQL += "\nCOMMENT ON INDEX items_code_idx IS 'issue 598 preserved comment';" + } + f := newIssue598Fixture(t, provider, indexSQL) + initial := f.generate(t) + require.Contains(t, initial.ToSQL(plan.SQLFormatRaw), "INDEX CONCURRENTLY", "fixture must exercise native online creation") + beforeRows := f.rows(t) + writer, err := f.db.BeginTx(context.Background(), nil) + require.NoError(t, err) + defer writer.Rollback() + _, err = writer.Exec("UPDATE items SET payload = payload WHERE id = 1") + require.NoError(t, err) + + result := make(chan error, 1) + config := f.savedApplyConfig(t, initial) + go func() { result <- apply.ApplyMigration(config, nil) }() + pid := f.waitForBuild(t, result) + + // While the invalid index belongs to a live build, a second plan must + // refuse to treat it as an abandoned index ready for recovery. + _, activeErr := planCmd.GeneratePlan(f.config, provider) + if assert.Error(t, activeErr, "planning must refuse an active concurrent index build") { + message := strings.ToLower(activeErr.Error()) + assert.True(t, strings.Contains(message, "active") || strings.Contains(message, "conflict") || strings.Contains(message, "wait"), "actionable active-build error: %v", activeErr) + } + + var cancelled bool + require.NoError(t, f.db.QueryRow("SELECT pg_cancel_backend($1)", pid).Scan(&cancelled)) + require.True(t, cancelled) + select { + case err := <-result: + require.Error(t, err, "interrupted concurrent create must report failure") + case <-time.After(10 * time.Second): + t.Fatal("native apply did not finish after cancelling its exact backend") + } + require.NoError(t, writer.Rollback()) + definition := f.assertIndex(t, false) + require.Equal(t, beforeRows, f.rows(t), "interruption must preserve every row") + + recovery := f.generate(t) + require.NotEmpty(t, recovery.Groups, "an abandoned invalid index must not be reported as converged") + require.Contains(t, recovery.ToSQL(plan.SQLFormatRaw), "REINDEX INDEX CONCURRENTLY", "recovery must retain the existing index until its replacement is ready") + require.NoError(t, f.apply(t, recovery)) + require.Equal(t, definition, f.assertIndex(t, true), "recovery must preserve the complete PostgreSQL index definition") + require.Equal(t, beforeRows, f.rows(t)) + if unique { + var comment string + require.NoError(t, f.db.QueryRow("SELECT obj_description('items_code_idx'::regclass, 'pg_class')").Scan(&comment)) + require.Equal(t, "issue 598 preserved comment", comment) + } + f.assertConstraints(t) + require.Empty(t, f.generate(t).Groups, "repeat native plan must converge only after the index is healthy") + }) + } +} + +// Failed uniqueness checks must remain visible and leave data correction to the +// operator. Only the explicit fixture UPDATE below changes application data. +func TestIssue598DuplicateDataRecovery(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + provider := testutil.SetupPostgres(t) + defer provider.Stop() + f := newIssue598Fixture(t, provider, "CREATE UNIQUE INDEX items_code_idx ON items (code);") + _, err := f.db.Exec("UPDATE items SET code = 1 WHERE id = 2") + require.NoError(t, err) + beforeRows := f.rows(t) + require.Error(t, f.apply(t, f.generate(t)), "duplicate data must fail native concurrent creation") + definition := f.assertIndex(t, false) + + // PostgreSQL must not allow an unusable unique index to become a constraint. + _, err = f.db.Exec("ALTER TABLE items ADD CONSTRAINT broken_unique UNIQUE USING INDEX items_code_idx") + require.Error(t, err) + require.Contains(t, strings.ToLower(err.Error()), "valid") + + recovery := f.generate(t) + require.NotEmpty(t, recovery.Groups, "failed unique creation must produce a recovery plan") + require.Contains(t, recovery.ToSQL(plan.SQLFormatRaw), "REINDEX INDEX CONCURRENTLY") + require.Error(t, f.apply(t, recovery), "recovery must fail while duplicates remain") + require.Equal(t, beforeRows, f.rows(t), "failed recovery must not remove or alter rows") + require.Equal(t, definition, f.assertIndex(t, false)) + f.assertConstraints(t) + + _, err = f.db.Exec("UPDATE items SET code = 2 WHERE id = 2") + require.NoError(t, err, "explicit fixture data correction") + correctedRows := f.rows(t) + require.NoError(t, f.apply(t, f.generate(t)), "a fresh plan must recover after explicit data correction") + require.Equal(t, definition, f.assertIndex(t, true)) + require.Equal(t, correctedRows, f.rows(t)) + f.assertConstraints(t) + require.Empty(t, f.generate(t).Groups) +} + +type issue598Fixture struct { + db *sql.DB + provider *postgres.EmbeddedPostgres + config *planCmd.PlanConfig +} + +func newIssue598Fixture(t *testing.T, provider *postgres.EmbeddedPostgres, indexSQL string) *issue598Fixture { + t.Helper() + target := testutil.SetupPostgres(t) + t.Cleanup(func() { target.Stop() }) + db, host, port, dbname, user, password := testutil.ConnectToPostgres(t, target) + t.Cleanup(func() { db.Close() }) + _, err := db.Exec(issue598Tables + ` +INSERT INTO parents VALUES (1); +INSERT INTO items VALUES (1, 1, 'first', 1), (2, 2, 'second', 1), (3, NULL, 'third', NULL);`) + require.NoError(t, err) + file := filepath.Join(t.TempDir(), "desired.sql") + require.NoError(t, os.WriteFile(file, []byte(issue598Tables+"\n"+indexSQL), 0644)) + return &issue598Fixture{db: db, provider: provider, config: &planCmd.PlanConfig{ + Host: host, Port: port, DB: dbname, User: user, Password: password, + Schema: "public", File: file, SSLMode: "disable", ApplicationName: "pgschema-issue598", + }} +} + +func (f *issue598Fixture) generate(t *testing.T) *plan.Plan { + t.Helper() + p, err := planCmd.GeneratePlan(f.config, f.provider) + require.NoError(t, err) + return p +} + +func (f *issue598Fixture) savedApplyConfig(t *testing.T, p *plan.Plan) *apply.ApplyConfig { + t.Helper() + data, err := json.Marshal(p) + require.NoError(t, err) + file := filepath.Join(t.TempDir(), "plan.json") + require.NoError(t, os.WriteFile(file, data, 0644)) + data, err = os.ReadFile(file) + require.NoError(t, err) + loaded, err := plan.FromJSON(data) // The same loader used by apply --plan. + require.NoError(t, err) + return &apply.ApplyConfig{ + Host: f.config.Host, Port: f.config.Port, DB: f.config.DB, + User: f.config.User, Password: f.config.Password, Schema: f.config.Schema, + Plan: loaded, SSLMode: "disable", ApplicationName: "pgschema-issue598-apply", + AutoApprove: true, Quiet: true, LockTimeout: "20s", + } +} + +func (f *issue598Fixture) apply(t *testing.T, p *plan.Plan) error { + t.Helper() + return apply.ApplyMigration(f.savedApplyConfig(t, p), nil) +} + +func (f *issue598Fixture) waitForBuild(t *testing.T, result <-chan error) int { + t.Helper() + deadline := time.After(10 * time.Second) + tick := time.NewTicker(20 * time.Millisecond) + defer tick.Stop() + for { + var pid int + err := f.db.QueryRow(` +SELECT p.pid +FROM pg_stat_progress_create_index p +JOIN pg_stat_activity a ON a.pid = p.pid +WHERE p.datid = (SELECT oid FROM pg_database WHERE datname = current_database()) + AND p.relid = 'public.items'::regclass + AND p.index_relid = to_regclass('public.items_code_idx') + AND p.command = 'CREATE INDEX CONCURRENTLY' + AND p.phase = 'waiting for writers before build' + AND a.application_name = 'pgschema-issue598-apply'`).Scan(&pid) + if err == nil { + return pid + } + require.ErrorIs(t, err, sql.ErrNoRows) + select { + case err := <-result: + t.Fatalf("apply ended before the concurrent build reached its writer wait: %v", err) + case <-deadline: + t.Fatal("concurrent build never reached the expected writer wait") + case <-tick.C: + } + } +} + +func (f *issue598Fixture) assertIndex(t *testing.T, wantValid bool) string { + t.Helper() + var valid, ready, live bool + var definition string + require.NoError(t, f.db.QueryRow(`SELECT indisvalid, indisready, indislive, pg_get_indexdef(indexrelid) +FROM pg_index WHERE indexrelid = 'public.items_code_idx'::regclass`).Scan(&valid, &ready, &live, &definition)) + require.Equal(t, wantValid, valid, "index validity") + require.True(t, live, "index must remain live") + if wantValid { + require.True(t, ready, "a recovered index must be ready for writes") + } + return definition +} + +func (f *issue598Fixture) rows(t *testing.T) string { + t.Helper() + var rows string + require.NoError(t, f.db.QueryRow("SELECT json_agg(items ORDER BY id)::text FROM items").Scan(&rows)) + return rows +} + +func (f *issue598Fixture) assertConstraints(t *testing.T) { + t.Helper() + var constraints, unhealthy int + require.NoError(t, f.db.QueryRow(`SELECT count(*), count(*) FILTER (WHERE NOT c.convalidated) +FROM pg_constraint c WHERE c.conrelid IN ('public.items'::regclass, 'public.parents'::regclass) +AND c.contype IN ('p', 'u', 'f')`).Scan(&constraints, &unhealthy)) + require.Equal(t, 4, constraints, "both primary keys, unique constraint and foreign key must survive") + require.Zero(t, unhealthy) + require.NoError(t, f.db.QueryRow(`SELECT count(*) FROM pg_index i JOIN pg_constraint c ON c.conindid = i.indexrelid +WHERE c.conrelid IN ('public.items'::regclass, 'public.parents'::regclass) +AND (NOT i.indisvalid OR NOT i.indisready OR NOT i.indislive)`).Scan(&unhealthy)) + require.Zero(t, unhealthy, "constraint backing indexes must stay usable") +} + +func TestIssue598PartitionStates(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + provider := testutil.SetupPostgres(t) + defer provider.Stop() + f := newIssue598Fixture(t, provider, "") + const partitionSQL = ` +CREATE TABLE partition_parent (id integer) PARTITION BY RANGE (id); +CREATE TABLE partition_child PARTITION OF partition_parent FOR VALUES FROM (0) TO (100); +CREATE INDEX parent_index ON ONLY partition_parent(id);` + _, err := f.db.Exec(partitionSQL) + require.NoError(t, err) + desired := issue598Tables + partitionSQL + require.NoError(t, os.WriteFile(f.config.File, []byte(desired), 0644)) + require.Empty(t, f.generate(t).Groups, "intentional invalid ON ONLY parent is not an abandoned concurrent build") + maintenance, err := f.db.Begin() + require.NoError(t, err) + defer maintenance.Rollback() + _, err = maintenance.Exec("LOCK TABLE ONLY partition_parent IN SHARE UPDATE EXCLUSIVE MODE") + require.NoError(t, err) + require.NoError(t, f.apply(t, f.generate(t)), "maintenance on an intentional invalid parent must allow a no-op") + desired += "\nALTER TABLE parents ADD COLUMN note text;" + require.NoError(t, os.WriteFile(f.config.File, []byte(desired), 0644)) + require.NoError(t, f.apply(t, f.generate(t)), "maintenance on an untouched invalid parent must not block unrelated DDL") + require.NoError(t, maintenance.Rollback()) + var valid, ready, live bool + require.NoError(t, f.db.QueryRow("SELECT indisvalid,indisready,indislive FROM pg_index WHERE indexrelid='parent_index'::regclass").Scan(&valid, &ready, &live)) + require.False(t, valid) + require.True(t, ready) + require.True(t, live) + require.NoError(t, os.WriteFile(f.config.File, []byte(strings.ReplaceAll(desired, "ON ONLY partition_parent", "ON partition_parent")), 0644)) + _, err = planCmd.GeneratePlan(f.config, provider) + require.ErrorContains(t, err, "attach", "an incomplete parent must not silently satisfy a desired valid parent") + require.NoError(t, f.db.QueryRow("SELECT indisvalid FROM pg_index WHERE indexrelid='parent_index'::regclass").Scan(&valid)) + require.False(t, valid) + + // Removing an intentionally incomplete parent needs no attachment repair. + withoutIndex := strings.ReplaceAll(desired, "CREATE INDEX parent_index ON ONLY partition_parent(id);", "") + require.NoError(t, os.WriteFile(f.config.File, []byte(withoutIndex), 0644)) + require.NoError(t, f.apply(t, f.generate(t))) + require.Empty(t, f.generate(t).Groups) + _, err = f.db.Exec("CREATE INDEX parent_index ON ONLY partition_parent(id)") + require.NoError(t, err) + // Recreate the same catalog state and remove the whole partitioned table. + require.NoError(t, os.WriteFile(f.config.File, []byte(issue598Tables+"\nALTER TABLE parents ADD COLUMN note text;"), 0644)) + require.NoError(t, f.apply(t, f.generate(t))) + var absent bool + require.NoError(t, f.db.QueryRow("SELECT to_regclass('partition_parent') IS NULL AND to_regclass('parent_index') IS NULL").Scan(&absent)) + require.True(t, absent) + require.Empty(t, f.generate(t).Groups) +} + +func TestIssue598QuotedIndexRecovery(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + provider := testutil.SetupPostgres(t) + defer provider.Stop() + const name = `odd'$pgschema$index` + f := newIssue598Fixture(t, provider, `CREATE UNIQUE INDEX "odd'$pgschema$index" ON items(code);`) + _, err := f.db.Exec("UPDATE items SET code=1 WHERE id=2") + require.NoError(t, err) + require.Error(t, f.apply(t, f.generate(t)), "produce a real invalid index with legal special characters in its name") + _, err = f.db.Exec("UPDATE items SET code=2 WHERE id=2") + require.NoError(t, err) + require.NoError(t, f.apply(t, f.generate(t)), "identifier text must not terminate generated assertion bodies") + var usable bool + require.NoError(t, f.db.QueryRow(`SELECT indisvalid AND indisready AND indislive FROM pg_index i JOIN pg_class c ON c.oid=i.indexrelid WHERE c.relname=$1`, name).Scan(&usable)) + require.True(t, usable) + require.Empty(t, f.generate(t).Groups) +} + +func TestIssue598PartitionConstraintRemoval(t *testing.T) { + if testing.Short() { + t.Skip("Skipping integration test in short mode") + } + provider := testutil.SetupPostgres(t) + defer provider.Stop() + f := newIssue598Fixture(t, provider, "") + const tables = `CREATE TABLE partition_parent (id integer) PARTITION BY RANGE (id); +CREATE TABLE partition_child PARTITION OF partition_parent FOR VALUES FROM (0) TO (100);` + _, err := f.db.Exec(tables + "ALTER TABLE ONLY partition_parent ADD CONSTRAINT parent_unique UNIQUE(id);") + require.NoError(t, err) + var valid bool + require.NoError(t, f.db.QueryRow("SELECT indisvalid FROM pg_index WHERE indexrelid='parent_unique'::regclass").Scan(&valid)) + require.False(t, valid) + require.NoError(t, os.WriteFile(f.config.File, []byte(issue598Tables+tables), 0644)) + removal := f.generate(t) + maintenance, err := f.db.Begin() + require.NoError(t, err) + defer maintenance.Rollback() + _, err = maintenance.Exec("LOCK TABLE ONLY partition_parent IN SHARE UPDATE EXCLUSIVE MODE") + require.NoError(t, err) + _, err = planCmd.GeneratePlan(f.config, provider) + require.ErrorContains(t, err, "maintenance") + require.ErrorContains(t, f.apply(t, removal), "maintenance", "saved constraint removal must recheck activity") + require.NoError(t, maintenance.Rollback()) + require.NoError(t, f.apply(t, f.generate(t))) + require.Empty(t, f.generate(t).Groups) + _, err = f.db.Exec("ALTER TABLE ONLY partition_parent ADD CONSTRAINT parent_unique UNIQUE(id)") + require.NoError(t, err) + require.NoError(t, os.WriteFile(f.config.File, []byte(issue598Tables), 0644)) + require.NoError(t, f.apply(t, f.generate(t))) + require.Empty(t, f.generate(t).Groups) + f.assertConstraints(t) +} diff --git a/cmd/plan/plan.go b/cmd/plan/plan.go index c409fbb63..a0dbc0117 100644 --- a/cmd/plan/plan.go +++ b/cmd/plan/plan.go @@ -405,8 +405,14 @@ func GeneratePlan(config *PlanConfig, provider postgres.DesiredStateProvider) (* fmt.Sscanf(v, "%d", &targetMajorVersion) } + // Repair abandoned concurrent builds before ordinary changes or cleanup. + repairs, err := diff.IndexRecoveryDiffs(currentStateIR, desiredStateIR, config.Schema) + if err != nil { + return nil, err + } + // Generate diff (current -> desired) using IR directly - diffs := diff.GenerateMigrationForTarget(currentStateIR, desiredStateIR, config.Schema, targetMajorVersion) + diffs := append(repairs, diff.GenerateMigrationForTarget(currentStateIR, desiredStateIR, config.Schema, targetMajorVersion)...) // Create plan from diffs with fingerprint migrationPlan := plan.NewPlanWithFingerprint(diffs, sourceFingerprint, targetMajorVersion, currentStateIR) diff --git a/docs/workflow/online-ddl.mdx b/docs/workflow/online-ddl.mdx index b75fdfba7..dc79370ea 100644 --- a/docs/workflow/online-ddl.mdx +++ b/docs/workflow/online-ddl.mdx @@ -189,3 +189,43 @@ ALTER TABLE orders VALIDATE CONSTRAINT orders_customer_fkey; ``` For more examples, see the test cases in [testdata/diff/online](https://github.com/pgplex/pgschema/tree/main/testdata/diff/online) which demonstrate online DDL patterns for various PostgreSQL schema operations. + +## Recovering interrupted concurrent index builds + +An interrupted `CREATE INDEX CONCURRENTLY` can leave a named but invalid index. +After the build has stopped, regenerate the plan against the same desired schema. +For an unchanged ordinary index definition, pgschema plans a schema-qualified +`REINDEX INDEX CONCURRENTLY` rather than reporting no changes. Apply the saved +plan normally. Recovery checks the catalog before and after rebuilding and keeps +PostgreSQL responsible for the definition, dependencies and uniqueness protection. +Both invalid indexes already accepting writes and those not yet ready are covered. + +Recovery runs outside a transaction, before ordinary migration steps. It is not +retried automatically within apply: after cancellation, lock timeout or another +failure, generate a fresh plan. Failed-reindex artifacts remain visible and use +ordinary dependency-respecting drop semantics when absent from the desired schema; +pgschema does not infer ownership from `_ccnew` or `_ccold` names. + +A unique index whose data contains duplicates still fails. Recovery does not alter +rows or remove the original index to force success. Resolve the data explicitly +and replan. The applying role needs PostgreSQL's required ownership or maintenance +privileges; pgschema does not escalate privileges. + +Planning refuses recovery of a desired invalid index whose definition is also +being changed. Recover the original definition first or use an explicitly authored +migration. Unhealthy required constraint-backed indexes and required indexes already being +dropped need explicit investigation. Integrity constraints are never removed as a +recovery strategy. + +An intentional invalid partition parent created with `CREATE INDEX ON ONLY` is +preserved when the desired schema describes the same state. If the desired parent +should be valid, attach the missing partition indexes explicitly; reindexing cannot +complete attachment. This distinction uses the inspected desired-state database. + +For affected unhealthy indexes, observed active builds or conflicting maintenance +locks cause a wait-and-replan error. The lock check also protects roles whose access +to another backend's progress details is restricted. It may temporarily refuse +recovery during maintenance such as VACUUM. Untouched intentional partition parents +do not block unrelated changes. These checks and the saved-plan fingerprint reduce +stale-state races; PostgreSQL's relation locks serialize the concurrent operation. +They do not provide a global lock against arbitrary external DDL. diff --git a/internal/diff/index_recovery.go b/internal/diff/index_recovery.go new file mode 100644 index 000000000..aef5c6365 --- /dev/null +++ b/internal/diff/index_recovery.go @@ -0,0 +1,107 @@ +package diff + +import ( + "fmt" + "sort" + + "github.com/pgplex/pgschema/ir" +) + +// IndexRecovery distinguishes repair from a definition change: the online +// replacement rewrite must never drop an invalid index before rebuilding it. +type IndexRecovery struct{ Index *ir.Index } + +func (r *IndexRecovery) GetObjectName() string { return r.Index.Name } + +// IndexRecoveryDiffs validates abnormal catalog states and returns repairs to +// execute before ordinary migration steps. PostgreSQL owns the replacement and +// dependency transfer; failed unique builds remain failures, never no-ops. +func IndexRecoveryDiffs(current, desired *ir.IR, schemaName string) ([]Diff, error) { + oldSchema, newSchema := current.Schemas[schemaName], desired.Schemas[schemaName] + if oldSchema == nil { + return nil, nil + } + names := make([]string, 0, len(oldSchema.IndexStates)) + for name := range oldSchema.IndexStates { + names = append(names, name) + } + sort.Strings(names) + var repairs []Diff + for _, name := range names { + state := oldSchema.IndexStates[name] + qualified := ir.QuoteIdentifier(schemaName) + "." + ir.QuoteIdentifier(name) + var wantedState *ir.IndexState + if newSchema != nil { + wantedState = newSchema.IndexStates[name] + } + // Explicit constraint removal is an ordinary dependency-aware migration, + // including an intentionally incomplete partitioned constraint. + if state.Constraint != "" && (newSchema == nil || newSchema.Tables[state.Table] == nil || newSchema.Tables[state.Table].Constraints[state.Constraint] == nil) { + if err := state.CheckOperation(schemaName, name); err != nil { + return nil, err + } + continue + } + if state.IndexKind == "I" { + // CREATE INDEX ON ONLY intentionally leaves a partitioned parent invalid. + // Attaching all matching leaf indexes, not REINDEX, validates that parent. + if wantedState != nil && wantedState.IndexKind == "I" && state.Table == wantedState.Table && state.Valid == wantedState.Valid && state.Ready == wantedState.Ready && state.Live == wantedState.Live && state.Constraint == wantedState.Constraint { + continue + } + if err := state.CheckOperation(schemaName, name); err != nil { + return nil, err + } + // An intentionally incomplete ordinary parent may be removed along + // with its index or table through the usual dependency-aware diff. + if state.Constraint == "" && findRecoveryIndex(newSchema, state.Table, name) == nil { + continue + } + return nil, fmt.Errorf("partitioned index %s is invalid; create and attach the missing partition indexes before replanning (REINDEX cannot complete partition attachment)", qualified) + } + if err := state.CheckOperation(schemaName, name); err != nil { + return nil, err + } + if state.Constraint != "" { + return nil, fmt.Errorf("constraint-backed index %s is not usable; inspect constraint %s and repair its index explicitly before replanning; automatic recovery will not drop integrity constraints", qualified, ir.QuoteIdentifier(state.Constraint)) + } + oldIndex := findRecoveryIndex(oldSchema, state.Table, name) + newIndex := findRecoveryIndex(newSchema, state.Table, name) + // Undesired physical indexes retain normal DROP semantics. This also permits + // cleanup of abandoned concurrent REINDEX artifacts, without name heuristics. + if newIndex == nil { + continue + } + if !state.Live || state.Valid || state.IndexKind != "i" || (state.TableKind != "r" && state.TableKind != "m") { + return nil, fmt.Errorf("index %s has unsupported catalog state (valid=%t, ready=%t, live=%t); finish or repair the interrupted operation before replanning", qualified, state.Valid, state.Ready, state.Live) + } + if wantedState != nil { + return nil, fmt.Errorf("desired index %s is not usable in the planning database; check the schema file and planning database before regenerating the plan", qualified) + } + if oldIndex == nil || !indexesStructurallyEqual(oldIndex, newIndex) { + return nil, fmt.Errorf("invalid index %s also has a requested definition change; recover its original definition first or use an explicitly authored migration, then regenerate the plan", qualified) + } + kind := DiffTypeTableIndex + if state.TableKind == "m" { + kind = DiffTypeMaterializedViewIndex + } + repairs = append(repairs, Diff{ + Type: kind, Operation: DiffOperationAlter, Path: schemaName + "." + state.Table + "." + name, + Source: &IndexRecovery{Index: oldIndex}, + Statements: []SQLStatement{{SQL: "REINDEX INDEX CONCURRENTLY " + qualified + ";", CanRunInTransaction: false}}, + }) + } + return repairs, nil +} + +func findRecoveryIndex(schema *ir.Schema, table, name string) *ir.Index { + if schema == nil { + return nil + } + if t := schema.Tables[table]; t != nil { + return t.Indexes[name] + } + if v := schema.Views[table]; v != nil { + return v.Indexes[name] + } + return nil +} diff --git a/internal/diff/index_recovery_test.go b/internal/diff/index_recovery_test.go new file mode 100644 index 000000000..3a43090ce --- /dev/null +++ b/internal/diff/index_recovery_test.go @@ -0,0 +1,78 @@ +package diff + +import ( + "testing" + + "github.com/pgplex/pgschema/ir" + "github.com/stretchr/testify/require" +) + +func TestIndexRecoveryClassification(t *testing.T) { + for _, tc := range []struct { + name string + mutate func(*ir.Schema, *ir.Schema) + wantError string + repairs int + }{ + {name: "unready", repairs: 1}, + {name: "ready_unique", repairs: 1, mutate: func(a, b *ir.Schema) { + a.IndexStates["idx"].Ready = true + a.Tables["t"].Indexes["idx"].Type = ir.IndexTypeUnique + b.Tables["t"].Indexes["idx"].Type = ir.IndexTypeUnique + }}, + {name: "comment_only", repairs: 1, mutate: func(a, b *ir.Schema) { b.Tables["t"].Indexes["idx"].Comment = "new comment" }}, + {name: "healthy", mutate: func(a, b *ir.Schema) { a.IndexStates = nil }}, + {name: "active", wantError: "active", mutate: func(a, b *ir.Schema) { a.IndexStates["idx"].BuildInProgress = true }}, + {name: "maintenance", wantError: "maintenance", mutate: func(a, b *ir.Schema) { a.IndexStates["idx"].ConflictingOperation = true }}, + {name: "dead_required", wantError: "unsupported catalog state", mutate: func(a, b *ir.Schema) { a.IndexStates["idx"].Live = false }}, + {name: "constraint", wantError: "constraint-backed", mutate: func(a, b *ir.Schema) { + a.IndexStates["idx"].Constraint = "unique_constraint" + b.Tables["t"].Constraints = map[string]*ir.Constraint{"unique_constraint": {Name: "unique_constraint"}} + }}, + {name: "undesired_constraint", mutate: func(a, b *ir.Schema) { a.IndexStates["idx"].Constraint = "removed_constraint" }}, + {name: "undesired_partition_constraint", mutate: func(a, b *ir.Schema) { + a.IndexStates["idx"].Constraint = "removed_constraint" + a.IndexStates["idx"].IndexKind = "I" + }}, + {name: "desired_unhealthy", wantError: "planning database", mutate: func(a, b *ir.Schema) { + v := *a.IndexStates["idx"] + b.IndexStates = map[string]*ir.IndexState{"idx": &v} + }}, + {name: "changed_unique_definition", wantError: "requested definition change", mutate: func(a, b *ir.Schema) { a.Tables["t"].Indexes["idx"].Type = ir.IndexTypeUnique }}, + {name: "undesired_invalid", mutate: func(a, b *ir.Schema) { delete(b.Tables["t"].Indexes, "idx") }}, + {name: "undesired_dead_cleanup", mutate: func(a, b *ir.Schema) { delete(b.Tables["t"].Indexes, "idx"); a.IndexStates["idx"].Live = false }}, + {name: "partition_index_removed", mutate: func(a, b *ir.Schema) { a.IndexStates["idx"].IndexKind = "I"; delete(b.Tables["t"].Indexes, "idx") }}, + {name: "partition_table_removed", mutate: func(a, b *ir.Schema) { a.IndexStates["idx"].IndexKind = "I"; delete(b.Tables, "t") }}, + {name: "partition_incomplete", wantError: "attach", mutate: func(a, b *ir.Schema) { a.IndexStates["idx"].IndexKind = "I" }}, + {name: "intentional_partition_with_maintenance", mutate: func(a, b *ir.Schema) { + a.IndexStates["idx"].IndexKind = "I" + a.IndexStates["idx"].ConflictingOperation = true + v := *a.IndexStates["idx"] + b.IndexStates = map[string]*ir.IndexState{"idx": &v} + }}, + {name: "intentional_partition", mutate: func(a, b *ir.Schema) { + a.IndexStates["idx"].IndexKind = "I" + v := *a.IndexStates["idx"] + b.IndexStates = map[string]*ir.IndexState{"idx": &v} + }}, + } { + t.Run(tc.name, func(t *testing.T) { + makeSchema := func() *ir.Schema { + return &ir.Schema{Name: "public", Tables: map[string]*ir.Table{"t": {Indexes: map[string]*ir.Index{"idx": {Name: "idx", Schema: "public", Table: "t", Type: ir.IndexTypeRegular, Method: "btree", Columns: []*ir.IndexColumn{{Name: "id", Position: 1}}}}}}} + } + a, b := makeSchema(), makeSchema() + a.IndexStates = map[string]*ir.IndexState{"idx": {Table: "t", IndexKind: "i", TableKind: "r", Live: true}} + if tc.mutate != nil { + tc.mutate(a, b) + } + repairs, err := IndexRecoveryDiffs(&ir.IR{Schemas: map[string]*ir.Schema{"public": a}}, &ir.IR{Schemas: map[string]*ir.Schema{"public": b}}, "public") + if tc.wantError != "" { + require.ErrorContains(t, err, tc.wantError) + require.Empty(t, repairs) + } else { + require.NoError(t, err) + require.Len(t, repairs, tc.repairs) + } + }) + } +} diff --git a/internal/fingerprint/index_state_test.go b/internal/fingerprint/index_state_test.go new file mode 100644 index 000000000..73050a3a4 --- /dev/null +++ b/internal/fingerprint/index_state_test.go @@ -0,0 +1,35 @@ +package fingerprint + +import ( + "encoding/json" + "testing" + + "github.com/pgplex/pgschema/ir" + "github.com/stretchr/testify/require" +) + +func TestIndexStateFingerprint(t *testing.T) { + schema := &ir.Schema{Name: "public"} + state := &ir.IR{Schemas: map[string]*ir.Schema{"public": schema}} + healthy, err := ComputeFingerprint(state, "public") + require.NoError(t, err) + original, err := json.Marshal(schema) + require.NoError(t, err) + require.NotContains(t, string(original), "index_states") + schema.IndexStates = map[string]*ir.IndexState{} + same, err := ComputeFingerprint(state, "public") + require.NoError(t, err) + require.Equal(t, healthy, same, "empty health metadata must preserve pre-fix fingerprints") + schema.IndexStates["idx"] = &ir.IndexState{Table: "t", IndexKind: "i", TableKind: "r", Live: true} + invalid, err := ComputeFingerprint(state, "public") + require.NoError(t, err) + require.NotEqual(t, healthy, invalid) + schema.IndexStates["idx"].BuildInProgress = true + active, err := ComputeFingerprint(state, "public") + require.NoError(t, err) + require.Equal(t, invalid, active, "PIDs/progress must not create schema drift") + schema.IndexStates["idx"].Ready = true + ready, err := ComputeFingerprint(state, "public") + require.NoError(t, err) + require.NotEqual(t, invalid, ready) +} diff --git a/internal/plan/index_recovery.go b/internal/plan/index_recovery.go new file mode 100644 index 000000000..914a7447c --- /dev/null +++ b/internal/plan/index_recovery.go @@ -0,0 +1,66 @@ +package plan + +import ( + "fmt" + "strings" + + "github.com/pgplex/pgschema/ir" +) + +// Concurrent reindexing keeps PostgreSQL in charge of index names, dependency +// transfer and uniqueness enforcement. Never route recovery through DROP/CREATE. +func generateIndexRecovery(index *ir.Index) []RewriteStep { + qualified := ir.QuoteIdentifier(index.Schema) + "." + ir.QuoteIdentifier(index.Name) + table := ir.QuoteIdentifier(index.Schema) + "." + ir.QuoteIdentifier(index.Table) + literal := func(s string) string { return "'" + strings.ReplaceAll(s, "'", "''") + "'" } + indexLiteral, tableLiteral := literal(qualified), literal(table) + failure := literal("index " + qualified + " changed or an index build/conflicting operation is active; wait for it to finish and regenerate the plan") + precondition := fmt.Sprintf(`BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_catalog.pg_index i + JOIN pg_catalog.pg_class c ON c.oid = i.indexrelid + WHERE i.indexrelid = pg_catalog.to_regclass(%s) + AND i.indrelid = pg_catalog.to_regclass(%s) + AND NOT i.indisvalid AND i.indislive AND c.relkind = 'i' + AND NOT EXISTS (SELECT 1 FROM pg_catalog.pg_constraint con + WHERE con.conindid = i.indexrelid AND con.contype IN ('p', 'u', 'x')) + ) OR EXISTS ( + SELECT 1 FROM pg_catalog.pg_stat_progress_create_index p + WHERE p.datid = (SELECT oid FROM pg_catalog.pg_database WHERE datname = current_database()) + AND p.relid = pg_catalog.to_regclass(%s) + ) OR EXISTS ( + SELECT 1 FROM pg_catalog.pg_locks l + WHERE l.database = (SELECT oid FROM pg_catalog.pg_database WHERE datname = current_database()) + AND l.relation = pg_catalog.to_regclass(%s) + AND l.mode = 'ShareUpdateExclusiveLock' AND l.granted + AND l.pid IS DISTINCT FROM pg_backend_pid() + ) THEN + RAISE EXCEPTION USING MESSAGE = %s; + END IF; +END`, indexLiteral, tableLiteral, tableLiteral, tableLiteral, failure) + postcondition := fmt.Sprintf(`BEGIN + IF NOT EXISTS ( + SELECT 1 FROM pg_catalog.pg_index i + WHERE i.indexrelid = pg_catalog.to_regclass(%s) + AND i.indrelid = pg_catalog.to_regclass(%s) + AND i.indisvalid AND i.indisready AND i.indislive + ) THEN + RAISE EXCEPTION USING MESSAGE = %s; + END IF; +END`, indexLiteral, tableLiteral, literal("index "+qualified+" is still not valid and ready after recovery; inspect PostgreSQL errors and regenerate the plan")) + return []RewriteStep{ + {SQL: recoveryAssertion(precondition), CanRunInTransaction: true, RequiresIsolation: true}, + {SQL: "REINDEX INDEX CONCURRENTLY " + qualified + ";", CanRunInTransaction: false}, + {SQL: recoveryAssertion(postcondition), CanRunInTransaction: true, RequiresIsolation: true}, + } +} + +// Choose a delimiter absent from the body, including quoted identifiers and +// error messages. A legal index name can itself contain "$pgschema$". +func recoveryAssertion(body string) string { + tag := "$pgschema$" + for strings.Contains(body, tag) { + tag = strings.TrimSuffix(tag, "$") + "_$" + } + return "DO " + tag + "\n" + body + "\n" + tag + ";" +} diff --git a/internal/plan/index_recovery_test.go b/internal/plan/index_recovery_test.go new file mode 100644 index 000000000..0b2b614d6 --- /dev/null +++ b/internal/plan/index_recovery_test.go @@ -0,0 +1,23 @@ +package plan + +import ( + "strings" + "testing" + + "github.com/pgplex/pgschema/internal/diff" + "github.com/pgplex/pgschema/ir" + "github.com/stretchr/testify/require" +) + +func TestIndexRecoveryIsolation(t *testing.T) { + index := &ir.Index{Schema: "Quoted.Schema", Table: "table", Name: "idx'$pgschema$"} + p := NewPlan([]diff.Diff{{Type: diff.DiffTypeTableIndex, Operation: diff.DiffOperationAlter, Source: &diff.IndexRecovery{Index: index}}, {Type: diff.DiffTypeTable, Operation: diff.DiffOperationAlter, Statements: []diff.SQLStatement{{SQL: "SELECT 1;", CanRunInTransaction: true}}}}, 18, nil) + require.Len(t, p.Groups, 4, "precondition, concurrent rebuild, postcondition and following DDL must be separate") + for _, g := range p.Groups { + require.Len(t, g.Steps, 1) + } + require.Equal(t, `REINDEX INDEX CONCURRENTLY "Quoted.Schema"."idx'$pgschema$";`, p.Groups[1].Steps[0].SQL) + require.True(t, strings.HasPrefix(p.Groups[0].Steps[0].SQL, "DO $pgschema_$"), "legal names must not terminate the DO body") + require.Contains(t, p.Groups[0].Steps[0].SQL, "idx''$pgschema$") + require.Contains(t, p.Groups[2].Steps[0].SQL, "RAISE EXCEPTION") +} diff --git a/internal/plan/rewrite.go b/internal/plan/rewrite.go index ccbd5bd7d..212b4357b 100644 --- a/internal/plan/rewrite.go +++ b/internal/plan/rewrite.go @@ -28,6 +28,10 @@ type RewriteStep struct { // (nil-safe); rewrites consult it to pick constraint names that don't collide // with existing constraints, including ones invisible to the IR. func generateRewrite(d diff.Diff, newlyCreatedTables map[string]bool, newlyCreatedMaterializedViews map[string]bool, targetMajorVersion int, currentIR *ir.IR) []RewriteStep { + if recovery, ok := d.Source.(*diff.IndexRecovery); ok { + return generateIndexRecovery(recovery.Index) + } + // Dispatch to specific rewrite generators based on diff type and source switch d.Type { case diff.DiffTypeTableIndex: diff --git a/ir/index_state.go b/ir/index_state.go new file mode 100644 index 000000000..33c2c6385 --- /dev/null +++ b/ir/index_state.go @@ -0,0 +1,61 @@ +package ir + +import ( + "context" + "fmt" +) + +// IndexState records operational state separately from the index definition. +// PostgreSQL can retain an invalid index after an interrupted concurrent build, +// including one that is ready for writes and still enforces uniqueness. +type IndexState struct { + Table string `json:"table"` + IndexKind string `json:"index_kind"` + TableKind string `json:"table_kind"` + Valid bool `json:"valid"` + Ready bool `json:"ready"` + Live bool `json:"live"` + Constraint string `json:"constraint,omitempty"` + // Progress and locks are transient observations, not schema fingerprints. + BuildInProgress bool `json:"-"` + ConflictingOperation bool `json:"-"` +} + +func (i *Inspector) buildIndexStates(ctx context.Context, result *IR, targetSchema string) error { + rows, err := i.queries.GetUnhealthyIndexesForSchema(ctx, targetSchema) + if err != nil { + return err + } + schema := result.Schemas[targetSchema] + if schema == nil { + return nil + } + for _, row := range rows { + if i.ignoreConfig != nil && (i.ignoreConfig.ShouldIgnoreIndex(row.IndexName) || (row.ConstraintName != "" && i.ignoreConfig.ShouldIgnoreConstraint(row.ConstraintName))) { + continue + } + // The regular inspector already applies table/view/extension ignore rules. + if schema.Tables[row.TableName] == nil && schema.Views[row.TableName] == nil { + continue + } + if schema.IndexStates == nil { + schema.IndexStates = make(map[string]*IndexState) + } + schema.IndexStates[row.IndexName] = &IndexState{ + Table: row.TableName, IndexKind: row.IndexKind, TableKind: row.TableKind, + Valid: row.IsValid, Ready: row.IsReady, Live: row.IsLive, Constraint: row.ConstraintName, + BuildInProgress: row.BuildInProgress, ConflictingOperation: row.ConflictingOperation, + } + } + return nil +} + +// CheckOperation refuses to mutate an unhealthy index while its table has an +// in-flight build or conflicting maintenance. Call only for affected indexes: +// an intentional ON ONLY partition parent must not block unrelated plans. +func (s *IndexState) CheckOperation(schema, name string) error { + if s.BuildInProgress || s.ConflictingOperation { + return fmt.Errorf("index %s.%s is not usable while an index build or conflicting maintenance operation is active; wait for it to finish, then regenerate the plan", QuoteIdentifier(schema), QuoteIdentifier(name)) + } + return nil +} diff --git a/ir/inspector.go b/ir/inspector.go index 6233a088f..0e19adf62 100644 --- a/ir/inspector.go +++ b/ir/inspector.go @@ -170,6 +170,10 @@ func (i *Inspector) BuildIR(ctx context.Context, targetSchema string) (*IR, erro return nil, fmt.Errorf("failed to build indexes: %w", err) } + if err := i.buildIndexStates(ctx, schema, targetSchema); err != nil { + return nil, fmt.Errorf("failed to inspect index state: %w", err) + } + // Normalize the IR normalizeIR(schema) diff --git a/ir/ir.go b/ir/ir.go index 6bb360dc9..40e18db4c 100644 --- a/ir/ir.go +++ b/ir/ir.go @@ -34,7 +34,10 @@ type Schema struct { Privileges []*Privilege `json:"privileges,omitempty"` // Explicit privilege grants on objects ColumnPrivileges []*ColumnPrivilege `json:"column_privileges,omitempty"` // Column-level privilege grants RevokedDefaultPrivileges []*RevokedDefaultPrivilege `json:"revoked_default_privileges,omitempty"` // Explicit revokes of default PUBLIC privileges - mu sync.RWMutex // Protects concurrent access to all maps + // IndexStates contains only unhealthy indexes, including constraint-backed ones. + // Healthy schemas retain their existing serialization and fingerprints. + IndexStates map[string]*IndexState `json:"index_states,omitempty"` + mu sync.RWMutex // Protects concurrent access to all maps } // LikeClause represents a LIKE clause in CREATE TABLE statement diff --git a/ir/queries/queries.sql b/ir/queries/queries.sql index 5f022cdcb..627a095ed 100644 --- a/ir/queries/queries.sql +++ b/ir/queries/queries.sql @@ -2264,3 +2264,44 @@ WHERE d.classid = 'pg_proc'::regclass AND d.refclassid = 'pg_proc'::regclass AND d.deptype = 'n' AND dependent_ns.nspname = $1; + +-- GetUnhealthyIndexesForSchema includes constraint-backed indexes, whose +-- definitions are otherwise represented by pg_constraint. Partitioned +-- parents can be intentionally invalid; retain catalog state rather +-- than treating every indisvalid=false index as an abandoned build. +-- name: GetUnhealthyIndexesForSchema :many +SELECT + i.relname AS index_name, + t.relname AS table_name, + i.relkind::text AS index_kind, + t.relkind::text AS table_kind, + idx.indisvalid AS is_valid, + idx.indisready AS is_ready, + idx.indislive AS is_live, + COALESCE(con.conname, '')::text AS constraint_name, + EXISTS ( + SELECT 1 FROM pg_catalog.pg_stat_progress_create_index p + WHERE p.datid = (SELECT oid FROM pg_catalog.pg_database WHERE datname = current_database()) + AND p.relid = t.oid + ) AS build_in_progress, + EXISTS ( + SELECT 1 FROM pg_catalog.pg_locks l + WHERE l.database = (SELECT oid FROM pg_catalog.pg_database WHERE datname = current_database()) + AND l.relation = t.oid AND l.mode = 'ShareUpdateExclusiveLock' + AND l.granted AND l.pid IS DISTINCT FROM pg_backend_pid() + ) AS conflicting_operation +FROM pg_catalog.pg_index idx +JOIN pg_catalog.pg_class i ON i.oid = idx.indexrelid +JOIN pg_catalog.pg_class t ON t.oid = idx.indrelid +JOIN pg_catalog.pg_namespace n ON n.oid = t.relnamespace +LEFT JOIN pg_catalog.pg_constraint con ON con.conindid = idx.indexrelid AND con.contype IN ('p', 'u', 'x') +WHERE n.nspname = $1 + AND (NOT idx.indisvalid OR NOT idx.indisready OR NOT idx.indislive) + AND NOT EXISTS ( + SELECT 1 FROM pg_catalog.pg_depend dep + WHERE dep.classid = 'pg_catalog.pg_class'::regclass + AND dep.objid IN (t.oid, i.oid) AND dep.objsubid = 0 + AND dep.refclassid = 'pg_catalog.pg_extension'::regclass + AND dep.refobjsubid = 0 AND dep.deptype = 'e' + ) +ORDER BY i.relname; diff --git a/ir/queries/queries.sql.go b/ir/queries/queries.sql.go index f1783968e..aa5c7a8f0 100644 --- a/ir/queries/queries.sql.go +++ b/ir/queries/queries.sql.go @@ -4049,6 +4049,95 @@ func (q *Queries) GetTypesForSchema(ctx context.Context, dollar_1 sql.NullString return items, nil } +const getUnhealthyIndexesForSchema = `-- name: GetUnhealthyIndexesForSchema :many +SELECT + i.relname AS index_name, + t.relname AS table_name, + i.relkind::text AS index_kind, + t.relkind::text AS table_kind, + idx.indisvalid AS is_valid, + idx.indisready AS is_ready, + idx.indislive AS is_live, + COALESCE(con.conname, '')::text AS constraint_name, + EXISTS ( + SELECT 1 FROM pg_catalog.pg_stat_progress_create_index p + WHERE p.datid = (SELECT oid FROM pg_catalog.pg_database WHERE datname = current_database()) + AND p.relid = t.oid + ) AS build_in_progress, + EXISTS ( + SELECT 1 FROM pg_catalog.pg_locks l + WHERE l.database = (SELECT oid FROM pg_catalog.pg_database WHERE datname = current_database()) + AND l.relation = t.oid AND l.mode = 'ShareUpdateExclusiveLock' + AND l.granted AND l.pid IS DISTINCT FROM pg_backend_pid() + ) AS conflicting_operation +FROM pg_catalog.pg_index idx +JOIN pg_catalog.pg_class i ON i.oid = idx.indexrelid +JOIN pg_catalog.pg_class t ON t.oid = idx.indrelid +JOIN pg_catalog.pg_namespace n ON n.oid = t.relnamespace +LEFT JOIN pg_catalog.pg_constraint con ON con.conindid = idx.indexrelid AND con.contype IN ('p', 'u', 'x') +WHERE n.nspname = $1 + AND (NOT idx.indisvalid OR NOT idx.indisready OR NOT idx.indislive) + AND NOT EXISTS ( + SELECT 1 FROM pg_catalog.pg_depend dep + WHERE dep.classid = 'pg_catalog.pg_class'::regclass + AND dep.objid IN (t.oid, i.oid) AND dep.objsubid = 0 + AND dep.refclassid = 'pg_catalog.pg_extension'::regclass + AND dep.refobjsubid = 0 AND dep.deptype = 'e' + ) +ORDER BY i.relname +` + +type GetUnhealthyIndexesForSchemaRow struct { + IndexName string `db:"index_name" json:"index_name"` + TableName string `db:"table_name" json:"table_name"` + IndexKind string `db:"index_kind" json:"index_kind"` + TableKind string `db:"table_kind" json:"table_kind"` + IsValid bool `db:"is_valid" json:"is_valid"` + IsReady bool `db:"is_ready" json:"is_ready"` + IsLive bool `db:"is_live" json:"is_live"` + ConstraintName string `db:"constraint_name" json:"constraint_name"` + BuildInProgress bool `db:"build_in_progress" json:"build_in_progress"` + ConflictingOperation bool `db:"conflicting_operation" json:"conflicting_operation"` +} + +// GetUnhealthyIndexesForSchema includes constraint-backed indexes, whose +// definitions are otherwise represented by pg_constraint. Partitioned +// parents can be intentionally invalid; retain catalog state rather +// than treating every indisvalid=false index as an abandoned build. +func (q *Queries) GetUnhealthyIndexesForSchema(ctx context.Context, nspname string) ([]GetUnhealthyIndexesForSchemaRow, error) { + rows, err := q.db.QueryContext(ctx, getUnhealthyIndexesForSchema, nspname) + if err != nil { + return nil, err + } + defer rows.Close() + var items []GetUnhealthyIndexesForSchemaRow + for rows.Next() { + var i GetUnhealthyIndexesForSchemaRow + if err := rows.Scan( + &i.IndexName, + &i.TableName, + &i.IndexKind, + &i.TableKind, + &i.IsValid, + &i.IsReady, + &i.IsLive, + &i.ConstraintName, + &i.BuildInProgress, + &i.ConflictingOperation, + ); err != nil { + return nil, err + } + items = append(items, i) + } + if err := rows.Close(); err != nil { + return nil, err + } + if err := rows.Err(); err != nil { + return nil, err + } + return items, nil +} + const getViewDependencies = `-- name: GetViewDependencies :many SELECT DISTINCT vtu.view_schema AS dependent_schema,