mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-09-11 08:07:15 +00:00
fix: give every table a primary key so the db can be logically replicated (#11036)
* fix: give every table a primary key so the db can be logically replicated Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017WS9iNfYQiLJNuBnxysBzi * fix: tighten replicability guard and trim migration comments Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017WS9iNfYQiLJNuBnxysBzi * fix: split deployment_metadata into its own primary-key migration Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017WS9iNfYQiLJNuBnxysBzi * docs: correct the partial-index predicate note after the deployment_metadata split Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_017WS9iNfYQiLJNuBnxysBzi --------- Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
co-authored by
Claude Opus 5
Claude Opus 5
parent
8bd144ccb1
commit
e62bfdcd8c
@@ -0,0 +1,9 @@
|
||||
-- Dropping the column drops the primary key and the identity sequence with it, and
|
||||
-- only marks the column dropped in the catalog rather than rewriting the table, so
|
||||
-- this takes the ACCESS EXCLUSIVE lock but not the time.
|
||||
|
||||
ALTER TABLE workspace_runnable_dependencies DROP COLUMN IF EXISTS id;
|
||||
ALTER TABLE dbt_node DROP COLUMN IF EXISTS id;
|
||||
ALTER TABLE dbt_edge DROP COLUMN IF EXISTS id;
|
||||
ALTER TABLE dbt_column_edge DROP COLUMN IF EXISTS id;
|
||||
ALTER TABLE dbt_graph_snapshot DROP COLUMN IF EXISTS id;
|
||||
@@ -0,0 +1,31 @@
|
||||
-- These five had neither a PRIMARY KEY nor an explicit REPLICA IDENTITY, which makes
|
||||
-- PostgreSQL reject UPDATE and DELETE on them under logical replication.
|
||||
-- `deployment_metadata` and `metrics` are the other two, one migration each after this.
|
||||
--
|
||||
-- The surrogate cannot be swapped for a natural key: every unique index on all five is
|
||||
-- PARTIAL, split on `script_hash IS NULL`, and a partial index cannot back a primary
|
||||
-- key. The partial uniques stay; they are what the ON CONFLICT clauses infer.
|
||||
--
|
||||
-- Each ALTER rewrites its table under ACCESS EXCLUSIVE and holds it unavailable for
|
||||
-- the rewrite. These five share a transaction because each is bounded by what a
|
||||
-- workspace holds rather than by how long it has run, so none can grow into the one
|
||||
-- that locks the rest; a transaction holds all its locks until it commits.
|
||||
--
|
||||
-- An instance that cannot afford that lock at startup can set REPLICA IDENTITY FULL
|
||||
-- on these tables instead, which unblocks replication by itself, and run these
|
||||
-- idempotent ALTERs in a maintenance window first.
|
||||
|
||||
ALTER TABLE workspace_runnable_dependencies
|
||||
ADD COLUMN IF NOT EXISTS id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY;
|
||||
|
||||
ALTER TABLE dbt_node
|
||||
ADD COLUMN IF NOT EXISTS id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY;
|
||||
|
||||
ALTER TABLE dbt_edge
|
||||
ADD COLUMN IF NOT EXISTS id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY;
|
||||
|
||||
ALTER TABLE dbt_column_edge
|
||||
ADD COLUMN IF NOT EXISTS id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY;
|
||||
|
||||
ALTER TABLE dbt_graph_snapshot
|
||||
ADD COLUMN IF NOT EXISTS id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY;
|
||||
@@ -0,0 +1 @@
|
||||
ALTER TABLE deployment_metadata DROP COLUMN IF EXISTS id;
|
||||
@@ -0,0 +1,13 @@
|
||||
-- The sixth of the seven; why any of them need a key is in
|
||||
-- 20260909052532_add_missing_primary_keys.
|
||||
--
|
||||
-- Kept out of that migration because it is the one table in the set with no retention
|
||||
-- sweep -- rows accumulate per deployed script hash, flow version and app version for
|
||||
-- the life of the instance -- so on an instance that upgrades after years of deploys
|
||||
-- its ACCESS EXCLUSIVE rewrite is the one that could hold the others locked.
|
||||
--
|
||||
-- No natural key: each row is a script, flow OR app deployment, and the three unique
|
||||
-- indexes are partial on exactly that split, so none of them covers every row.
|
||||
|
||||
ALTER TABLE deployment_metadata
|
||||
ADD COLUMN IF NOT EXISTS id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY;
|
||||
@@ -0,0 +1 @@
|
||||
ALTER TABLE metrics DROP COLUMN IF EXISTS row_id;
|
||||
@@ -0,0 +1,12 @@
|
||||
-- The last of the seven; why any of them need a key is in
|
||||
-- 20260909052532_add_missing_primary_keys.
|
||||
--
|
||||
-- Kept out of that migration because it is the largest (~400 MB / 750k rows on the
|
||||
-- instance this was measured on, a steady state: `queue_%` rows, which are nearly all
|
||||
-- of them, are swept at 14 days) and its ALTER rewrites it under ACCESS EXCLUSIVE. One
|
||||
-- migration is one transaction, so alone it holds no lock on the others as it rewrites.
|
||||
--
|
||||
-- The surrogate cannot be called `id` -- `metrics.id` holds the metric NAME.
|
||||
|
||||
ALTER TABLE metrics
|
||||
ADD COLUMN IF NOT EXISTS row_id BIGINT GENERATED ALWAYS AS IDENTITY PRIMARY KEY;
|
||||
@@ -70,15 +70,15 @@ ci_test_reference: workspace_id(char), test_script_path(char), test_script_hash(
|
||||
concurrency_settings: hash(bigint), concurrency_key(char), concurrent_limit(int), concurrency_time_window_s(int)
|
||||
config: name(char), config(jsonb)
|
||||
custom_concurrency_key_ended: key(char), ended_at(ts)
|
||||
dbt_column_edge: workspace_id(char), script_path(char), script_hash(bigint), job_id(uuid), parent_unique_id(text), parent_column(text), child_unique_id(text), child_column(text), lineage_kind(text), ingested_at(ts)
|
||||
dbt_column_edge: workspace_id(char), script_path(char), script_hash(bigint), job_id(uuid), parent_unique_id(text), parent_column(text), child_unique_id(text), child_column(text), lineage_kind(text), ingested_at(ts), id(bigint)
|
||||
FK: (workspace_id) -> workspace(id), (workspace_id, script_hash) -> script(workspace_id, hash)
|
||||
dbt_edge: workspace_id(char), script_path(char), script_hash(bigint), job_id(uuid), parent_unique_id(text), child_unique_id(text), ingested_at(ts)
|
||||
dbt_edge: workspace_id(char), script_path(char), script_hash(bigint), job_id(uuid), parent_unique_id(text), child_unique_id(text), ingested_at(ts), id(bigint)
|
||||
FK: (workspace_id) -> workspace(id), (workspace_id, script_hash) -> script(workspace_id, hash)
|
||||
dbt_graph_snapshot: workspace_id(char), script_path(char), script_hash(bigint), job_id(uuid), digest(text), relation_root_at_last_ingest(text), ingested_at(ts), permissioned_as(char)
|
||||
dbt_graph_snapshot: workspace_id(char), script_path(char), script_hash(bigint), job_id(uuid), digest(text), relation_root_at_last_ingest(text), ingested_at(ts), permissioned_as(char), id(bigint)
|
||||
FK: (workspace_id) -> workspace(id), (workspace_id, script_hash) -> script(workspace_id, hash)
|
||||
dbt_environment_state: workspace_id(char), script_path(char), environment(text), job_id(uuid), manifest(text), manifest_key(text), run_results(text), run_results_key(text), updated_at(ts)
|
||||
FK: (workspace_id) -> workspace(id)
|
||||
dbt_node: workspace_id(char), script_path(char), script_hash(bigint), job_id(uuid), unique_id(text), resource_type(text), name(text), asset_path(text), materialized(text), materialize_strategy(text), unique_key(text), tags(text[]), description(text), test_kind(text), test_column(text), test_args(jsonb), severity(text), attached_node(text), columns(jsonb), column_schema(jsonb), freshness(jsonb), raw_code(text), original_file_path(text), ingested_at(ts)
|
||||
dbt_node: workspace_id(char), script_path(char), script_hash(bigint), job_id(uuid), unique_id(text), resource_type(text), name(text), asset_path(text), materialized(text), materialize_strategy(text), unique_key(text), tags(text[]), description(text), test_kind(text), test_column(text), test_args(jsonb), severity(text), attached_node(text), columns(jsonb), column_schema(jsonb), freshness(jsonb), raw_code(text), original_file_path(text), ingested_at(ts), id(bigint)
|
||||
FK: (workspace_id) -> workspace(id), (workspace_id, script_hash) -> script(workspace_id, hash)
|
||||
dbt_run_progress: workspace_id(char), job_id(uuid), asset_kind(asset_kind), asset_path(char), status(materialization_status), row_count(bigint), error(text), updated_at(ts)
|
||||
FK: (workspace_id) -> workspace(id)
|
||||
@@ -88,7 +88,7 @@ debounce_key: key(char), job_id(uuid), previous_job_id(uuid), first_started_at(t
|
||||
debounce_stale_data: job_id(uuid), to_relock(text[])
|
||||
debouncing_settings: hash(bigint), debounce_key(char), debounce_delay_s(int), max_total_debouncing_time(int), max_total_debounces_amount(int), debounce_args_to_accumulate(text[])
|
||||
dependency_map: workspace_id(char), importer_path(char), importer_kind(importer_kind), imported_path(char), importer_node_id(char)
|
||||
deployment_metadata: workspace_id(char), path(char), script_hash(bigint), app_version(bigint), callback_job_ids(uuid[]), deployment_msg(text), flow_version(bigint), job_id(uuid)
|
||||
deployment_metadata: workspace_id(char), path(char), script_hash(bigint), app_version(bigint), callback_job_ids(uuid[]), deployment_msg(text), flow_version(bigint), job_id(uuid), id(bigint)
|
||||
FK: (workspace_id) -> workspace(id)
|
||||
draft: workspace_id(char), path(char), typ(draft_type), value(json), created_at(ts)
|
||||
FK: (workspace_id) -> workspace(id)
|
||||
@@ -147,7 +147,7 @@ mcp_oauth_refresh_token: id(bigint), refresh_token(char), access_token_hash(char
|
||||
mcp_oauth_server_client: client_id(char), client_name(char), redirect_uris(text[]), created_at(ts)
|
||||
mcp_oauth_server_code: code(char), client_id(char), user_email(char), workspace_id(char), scopes(text[]), redirect_uri(text), code_challenge(char), code_challenge_method(char), created_at(ts), expires_at(ts)
|
||||
FK: (client_id) -> mcp_oauth_server_client(client_id)
|
||||
metrics: id(char), value(jsonb), created_at(ts)
|
||||
metrics: id(char), value(jsonb), created_at(ts), row_id(bigint)
|
||||
mqtt_trigger: mqtt_resource_path(char), subscribe_topics(jsonb[]), client_version(mqtt_client_version), v5_config(jsonb), v3_config(jsonb), client_id(char), path(char), script_path(char), is_flow(bool), workspace_id(char), edited_by(char), email(char), edited_at(ts), extra_perms(jsonb), server_id(char), last_server_ping(ts), error(text), error_handler_path(char), error_handler_args(jsonb), retry(jsonb), mode(trigger_mode), labels(text[])
|
||||
native_trigger: external_id(char), workspace_id(char), service_name(native_trigger_service), script_path(char), is_flow(bool), webhook_token_hash(char), service_config(jsonb), error(text), created_at(ts), updated_at(ts), enabled(bool)
|
||||
FK: (workspace_id) -> workspace(id)
|
||||
@@ -225,7 +225,7 @@ workspace_key: workspace_id(char), kind(workspace_key_kind), key(char)
|
||||
FK: (workspace_id) -> workspace(id)
|
||||
workspace_protection_rule: workspace_id(char), name(char), rules(int), bypass_groups(text[]), bypass_users(text[]), created_at(ts)
|
||||
FK: (workspace_id) -> workspace(id)
|
||||
workspace_runnable_dependencies: flow_path(char), runnable_path(char), script_hash(bigint), runnable_is_flow(bool), workspace_id(char), app_path(char)
|
||||
workspace_runnable_dependencies: flow_path(char), runnable_path(char), script_hash(bigint), runnable_is_flow(bool), workspace_id(char), app_path(char), id(bigint)
|
||||
FK: (app_path, workspace_id) -> app(path, workspace_id) | (flow_path, workspace_id) -> flow(path, workspace_id)
|
||||
workspace_settings: workspace_id(char), slack_team_id(char), slack_name(char), slack_command_script(char), slack_email(char), customer_id(char), plan(char), webhook(text), ai_config(jsonb), large_file_storage(jsonb), git_sync(jsonb), default_app(char), default_scripts(jsonb), deploy_ui(jsonb), mute_critical_alerts(bool), color(char), operator_settings(jsonb), teams_command_script(text), teams_team_id(text), teams_team_name(text), git_app_installations(jsonb), ducklake(jsonb), slack_oauth_client_id(char), slack_oauth_client_secret(char), datatable(jsonb), teams_team_guid(text), auto_invite(jsonb), error_handler(jsonb), success_handler(jsonb), public_app_execution_limit_per_minute(int), dbt_warehouses(jsonb), guest_access_enabled(bool), guest_jwt_public_key(text), guest_jwt_jwks_url(text)
|
||||
FK: (workspace_id) -> workspace(id)
|
||||
|
||||
@@ -0,0 +1,46 @@
|
||||
//! Every table must be replicable.
|
||||
//!
|
||||
//! PostgreSQL refuses UPDATE and DELETE on a table that has neither a PRIMARY KEY
|
||||
//! nor an explicit REPLICA IDENTITY once the database is published to a logical
|
||||
//! replication slot. That is what a low-downtime major-version upgrade runs on
|
||||
//! (RDS and Aurora Blue/Green, pglogical) and what every CDC pipeline reads, so a
|
||||
//! single keyless table blocks the upgrade outright. This runs against a freshly
|
||||
//! migrated database and fails on the migration that introduces one.
|
||||
|
||||
use sqlx::{Pool, Postgres};
|
||||
|
||||
/// Partitioned parents are checked alongside ordinary tables: a parent without a
|
||||
/// key hands the same defect to every partition created under it later.
|
||||
#[sqlx::test(migrations = "../migrations")]
|
||||
async fn every_table_is_replicable(db: Pool<Postgres>) -> anyhow::Result<()> {
|
||||
let offenders: Vec<String> = sqlx::query_scalar(
|
||||
"SELECT n.nspname || '.' || c.relname
|
||||
FROM pg_class c
|
||||
JOIN pg_namespace n ON n.oid = c.relnamespace
|
||||
WHERE c.relkind IN ('r', 'p')
|
||||
AND n.nspname NOT IN ('pg_catalog', 'information_schema')
|
||||
AND NOT (
|
||||
-- FULL and USING INDEX replicate on their own.
|
||||
c.relreplident IN ('f', 'i')
|
||||
-- DEFAULT resolves to the primary key, so it needs one to exist.
|
||||
-- NOTHING never replicates, primary key or not.
|
||||
OR (c.relreplident = 'd' AND EXISTS (
|
||||
SELECT 1 FROM pg_index i WHERE i.indrelid = c.oid AND i.indisprimary
|
||||
))
|
||||
)
|
||||
ORDER BY 1",
|
||||
)
|
||||
.fetch_all(&db)
|
||||
.await?;
|
||||
|
||||
assert!(
|
||||
offenders.is_empty(),
|
||||
"logical replication will reject UPDATE and DELETE on these tables, because \
|
||||
they carry no replica identity it can use: {}. \
|
||||
Give each one a primary key -- a natural composite key where every column \
|
||||
is NOT NULL, otherwise a surrogate `BIGINT GENERATED ALWAYS AS IDENTITY`.",
|
||||
offenders.join(", ")
|
||||
);
|
||||
|
||||
Ok(())
|
||||
}
|
||||
Reference in New Issue
Block a user