airajena commented on code in PR #102: URL: https://github.com/apache/fineract-business-intelligence/pull/102#discussion_r4380524559
########## warehouse/schema/platform.sql: ########## @@ -0,0 +1,1139 @@ +-- Licensed to the Apache Software Foundation (ASF) under one or more +-- contributor license agreements. See the NOTICE file distributed with +-- this work for additional information regarding copyright ownership. +-- The ASF licenses this file to You under the Apache License, Version 2.0 +-- (the "License"); you may not use this file except in compliance with +-- the License. You may obtain a copy of the License at +-- +-- http://www.apache.org/licenses/LICENSE-2.0 +-- +-- Unless required by applicable law or agreed to in writing, software +-- distributed under the License is distributed on an "AS IS" BASIS, +-- WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +-- See the License for the specific language governing permissions and +-- limitations under the License. + +CREATE EXTENSION IF NOT EXISTS pgcrypto; + +CREATE SCHEMA IF NOT EXISTS raw; +CREATE SCHEMA IF NOT EXISTS raw_shadow; +CREATE SCHEMA IF NOT EXISTS raw_views; +CREATE SCHEMA IF NOT EXISTS staging; +CREATE SCHEMA IF NOT EXISTS intermediate; +CREATE SCHEMA IF NOT EXISTS analytics; +CREATE SCHEMA IF NOT EXISTS meta; + +CREATE TABLE IF NOT EXISTS meta.platform_settings ( + setting_key TEXT PRIMARY KEY, + setting_value TEXT NOT NULL, + migration_sensitive BOOLEAN NOT NULL DEFAULT FALSE, + description TEXT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +INSERT INTO meta.platform_settings (setting_key, setting_value, migration_sensitive, description) +VALUES + ('raw_hash_partitions', '8', TRUE, + 'Hash partitions per raw table, keyed by source_id. Changing requires a re-partition migration.'), + ('raw_fillfactor', '85', FALSE, + 'Fillfactor for new raw partitions; below 100 keeps HOT updates for frequently updated rows.'), + ('require_dual_control', 'true', FALSE, + 'When true a cutover must be approved by a different actor than the requester.'), + ('dual_control_mode', 'parameter', FALSE, + 'parameter: approver is a CLI/DAG parameter (development). database: the approver must be a different ' + 'database login that recorded the approval through meta.approve_cutover().'), + ('reconciliation_max_age_minutes', '120', FALSE, + 'A cutover needs a PASSED shadow_vs_prod reconciliation for the same source and table newer than this.'), + ('change_log_retention_days', '14', FALSE, + 'Consumed entity_change_log partitions older than this are dropped by maintenance.'), + ('cursor_stale_days', '7', FALSE, + 'A dbt change cursor that has not moved for this long no longer pins change-log retention.'), + ('progress_retention_days', '7', FALSE, + 'ingestion_progress and ingestion_audit rows older than this are purged by maintenance.'), + ('ownership_audit_retention_days', '730', FALSE, + 'ingestion_ownership_audit rows older than this are purged by meta.purge_ownership_audit().') +ON CONFLICT (setting_key) DO NOTHING; + +CREATE OR REPLACE FUNCTION meta.setting(p_key TEXT, p_default TEXT DEFAULT NULL) +RETURNS TEXT +LANGUAGE sql +STABLE +AS $$ + SELECT COALESCE((SELECT setting_value FROM meta.platform_settings WHERE setting_key = p_key), p_default) +$$; + +CREATE TABLE IF NOT EXISTS meta.table_contract ( + table_name TEXT PRIMARY KEY, + raw_table TEXT NOT NULL UNIQUE, + primary_key TEXT NOT NULL, + columns JSONB NOT NULL, + indexes JSONB NOT NULL DEFAULT '[]'::jsonb, + change_entity TEXT CHECK (change_entity IN ('loan', 'client', 'global')), + change_key TEXT, + change_date TEXT, + contract_version INTEGER NOT NULL, + loaded_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +ALTER TABLE meta.table_contract ADD COLUMN IF NOT EXISTS change_history_columns JSONB; + +CREATE OR REPLACE FUNCTION meta.assert_identifier(p_value TEXT) +RETURNS TEXT +LANGUAGE plpgsql +IMMUTABLE +AS $$ +BEGIN + IF p_value IS NULL OR p_value !~ '^[a-z_][a-z0-9_]{0,62}$' THEN + RAISE EXCEPTION 'Invalid identifier in table contract: %', p_value USING ERRCODE = '22023'; + END IF; + RETURN p_value; +END; +$$; + +CREATE OR REPLACE FUNCTION meta.load_table_contract(p_contract JSONB) +RETURNS INTEGER +LANGUAGE plpgsql +AS $$ +DECLARE + t JSONB; + col JSONB; + n INTEGER := 0; +BEGIN + FOR t IN SELECT value FROM jsonb_array_elements(p_contract -> 'tables') LOOP + PERFORM meta.assert_identifier(t ->> 'table_name'); + PERFORM meta.assert_identifier(t ->> 'raw_table'); + PERFORM meta.assert_identifier(t ->> 'primary_key'); + FOR col IN SELECT value FROM jsonb_array_elements(t -> 'columns') LOOP + PERFORM meta.assert_identifier(col ->> 'name'); + IF (col ->> 'type') !~ '^(bigint|integer|smallint|text|date|timestamptz|boolean|numeric\(\d{1,2},\d{1,2}\))$' THEN + RAISE EXCEPTION 'Unsupported column type % for %.%', + col ->> 'type', t ->> 'table_name', col ->> 'name' USING ERRCODE = '22023'; + END IF; + END LOOP; + IF t -> 'change_tracking' -> 'history_columns' IS NOT NULL THEN + FOR col IN SELECT value FROM jsonb_array_elements(t -> 'change_tracking' -> 'history_columns') LOOP + PERFORM meta.assert_identifier(col #>> '{}'); + IF NOT EXISTS ( + SELECT 1 FROM jsonb_array_elements(t -> 'columns') c WHERE c ->> 'name' = col #>> '{}' + ) THEN + RAISE EXCEPTION 'History column % is not a column of %', col #>> '{}', t ->> 'table_name'; + END IF; + END LOOP; + END IF; + IF NOT EXISTS ( + SELECT 1 FROM jsonb_array_elements(t -> 'columns') c WHERE c ->> 'name' = t ->> 'primary_key' + ) THEN + RAISE EXCEPTION 'Primary key % missing from columns of %', t ->> 'primary_key', t ->> 'table_name'; + END IF; + + INSERT INTO meta.table_contract ( + table_name, raw_table, primary_key, columns, indexes, + change_entity, change_key, change_date, change_history_columns, contract_version + ) + VALUES ( + t ->> 'table_name', + t ->> 'raw_table', + t ->> 'primary_key', + t -> 'columns', + COALESCE(t -> 'indexes', '[]'::jsonb), + t -> 'change_tracking' ->> 'entity', + meta.assert_identifier(COALESCE(t -> 'change_tracking' ->> 'key', 'id')), + CASE WHEN t -> 'change_tracking' ->> 'date' IS NULL THEN NULL + ELSE meta.assert_identifier(t -> 'change_tracking' ->> 'date') END, + t -> 'change_tracking' -> 'history_columns', + (p_contract ->> 'contract_version')::INTEGER + ) + ON CONFLICT (table_name) DO UPDATE + SET raw_table = EXCLUDED.raw_table, + primary_key = EXCLUDED.primary_key, + columns = EXCLUDED.columns, + indexes = EXCLUDED.indexes, + change_entity = EXCLUDED.change_entity, + change_key = EXCLUDED.change_key, + change_date = EXCLUDED.change_date, + change_history_columns = EXCLUDED.change_history_columns, + contract_version = EXCLUDED.contract_version, + loaded_at = NOW(); + n := n + 1; + END LOOP; + RETURN n; +END; +$$; + +CREATE OR REPLACE FUNCTION meta.provision_raw_tables() +RETURNS VOID +LANGUAGE plpgsql +AS $$ +DECLARE + c RECORD; + s TEXT; + col JSONB; + idx JSONB; + ddl TEXT; + idx_cols TEXT; + idx_name TEXT; + parts INTEGER := meta.setting('raw_hash_partitions', '8')::INTEGER; + ff INTEGER := meta.setting('raw_fillfactor', '85')::INTEGER; + i INTEGER; +BEGIN + FOR c IN SELECT * FROM meta.table_contract ORDER BY table_name LOOP + FOREACH s IN ARRAY ARRAY['raw', 'raw_shadow'] LOOP + IF to_regclass(format('%I.%I', s, c.raw_table)) IS NULL THEN + ddl := format('CREATE TABLE %I.%I (source_id TEXT NOT NULL, tenant_id TEXT NOT NULL', s, c.raw_table); + FOR col IN SELECT value FROM jsonb_array_elements(c.columns) LOOP + ddl := ddl || format(', %I %s', col ->> 'name', col ->> 'type'); + END LOOP; + ddl := ddl + || ', _source_generation INTEGER NOT NULL DEFAULT 0' + || ', _source_lsn BIGINT' + || ', _source_ts TIMESTAMPTZ' + || ', _ingested_at TIMESTAMPTZ NOT NULL DEFAULT NOW()' + || ', _ingest_engine TEXT NOT NULL' + || ', _is_deleted BOOLEAN NOT NULL DEFAULT FALSE' + || format(', PRIMARY KEY (source_id, tenant_id, %I)) PARTITION BY HASH (source_id)', c.primary_key); + EXECUTE ddl; + FOR i IN 0 .. parts - 1 LOOP + EXECUTE format( + 'CREATE TABLE %I.%I PARTITION OF %I.%I FOR VALUES WITH (MODULUS %s, REMAINDER %s) WITH (fillfactor = %s)', + s, c.raw_table || '_p' || lpad(i::TEXT, 3, '0'), s, c.raw_table, parts, i, ff + ); + END LOOP; + ELSE + FOR col IN SELECT value FROM jsonb_array_elements(c.columns) LOOP + EXECUTE format('ALTER TABLE %I.%I ADD COLUMN IF NOT EXISTS %I %s', + s, c.raw_table, col ->> 'name', col ->> 'type'); + END LOOP; + END IF; + + EXECUTE format('CREATE STATISTICS IF NOT EXISTS %I.%I (dependencies, ndistinct) ON source_id, tenant_id, %I FROM %I.%I', + s, left('st_' || c.raw_table || '_identity', 63), c.primary_key, s, c.raw_table); + + FOR idx IN SELECT value FROM jsonb_array_elements(c.indexes) LOOP + SELECT string_agg(format('%I', meta.assert_identifier(v)), ', '), + string_agg(v, '_') + INTO idx_cols, idx_name + FROM jsonb_array_elements_text(idx) AS v; + idx_name := left('ix_' || c.raw_table || '_' || idx_name, 63); + EXECUTE format('CREATE INDEX IF NOT EXISTS %I ON %I.%I (source_id, tenant_id, %s)', + idx_name, s, c.raw_table, idx_cols); + END LOOP; + END LOOP; + END LOOP; +END; +$$; + +CREATE TABLE IF NOT EXISTS meta.source_registry ( + source_id TEXT PRIMARY KEY CHECK (source_id ~ '^[a-z0-9][a-z0-9_-]{0,62}$'), + tenant_id TEXT NOT NULL, + display_name TEXT, + region TEXT NOT NULL DEFAULT 'default', + tier TEXT NOT NULL DEFAULT 'standard', + db_host TEXT, + db_port INTEGER NOT NULL DEFAULT 5432, + db_name TEXT, + db_schema TEXT NOT NULL DEFAULT 'public', + compat_schema TEXT NOT NULL DEFAULT 'bi_connector_source', + db_user TEXT, + secret_ref TEXT, + connect_cluster TEXT NOT NULL DEFAULT 'default', + lifecycle_state TEXT NOT NULL DEFAULT 'REQUESTED' CHECK (lifecycle_state IN ( + 'REQUESTED', 'VALIDATING', 'PROVISIONING', 'ACTIVE', + 'PAUSED', 'DEGRADED', 'RETIRING', 'RETIRED' + )), + capture_generation INTEGER NOT NULL DEFAULT 0, + cob_job_name TEXT NOT NULL DEFAULT 'LOAN_COB', + labels JSONB NOT NULL DEFAULT '{}'::jsonb, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + CONSTRAINT secret_ref_is_reference CHECK (secret_ref IS NULL OR secret_ref ~ '^(env|file):') +); + +ALTER TABLE meta.source_registry DROP CONSTRAINT IF EXISTS secret_ref_is_reference; +ALTER TABLE meta.source_registry ADD CONSTRAINT secret_ref_is_reference + CHECK (secret_ref IS NULL OR secret_ref ~ '^(env|file):') NOT VALID; + +CREATE INDEX IF NOT EXISTS ix_source_registry_region ON meta.source_registry (region, lifecycle_state); + +CREATE TABLE IF NOT EXISTS meta.source_generation_floor ( + source_id TEXT PRIMARY KEY, + next_generation INTEGER NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE TABLE IF NOT EXISTS meta.ingestion_ownership ( + source_id TEXT NOT NULL REFERENCES meta.source_registry (source_id) ON DELETE CASCADE, + table_name TEXT NOT NULL REFERENCES meta.table_contract (table_name), + engine TEXT NOT NULL CHECK (engine IN ('PYTHON', 'KAFKA', 'NONE')), + shadow_engine TEXT CHECK (shadow_engine IN ('PYTHON', 'KAFKA')), + cutover_state TEXT NOT NULL DEFAULT 'STEADY' CHECK (cutover_state IN ( + 'STEADY', 'SHADOWING', 'RECONCILING', 'CUTOVER_READY', 'PAUSED' + )), + epoch BIGINT NOT NULL DEFAULT 1, + last_reconciliation_id BIGINT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + updated_by TEXT NOT NULL DEFAULT 'system', + change_reason TEXT, + PRIMARY KEY (source_id, table_name), + CONSTRAINT shadow_differs_from_engine CHECK (shadow_engine IS NULL OR shadow_engine <> engine), + CONSTRAINT shadow_state_consistent CHECK ( + (shadow_engine IS NULL AND cutover_state IN ('STEADY', 'PAUSED')) + OR (shadow_engine IS NOT NULL AND cutover_state IN ('SHADOWING', 'RECONCILING', 'CUTOVER_READY', 'PAUSED')) + ) +); + +ALTER TABLE meta.ingestion_ownership ADD COLUMN IF NOT EXISTS shadow_requested_by TEXT; +ALTER TABLE meta.ingestion_ownership ADD COLUMN IF NOT EXISTS shadow_requested_db_user TEXT; + +CREATE INDEX IF NOT EXISTS ix_ingestion_ownership_engine ON meta.ingestion_ownership (engine, shadow_engine); + +CREATE TABLE IF NOT EXISTS meta.ingestion_ownership_audit ( + audit_id BIGSERIAL PRIMARY KEY, + source_id TEXT NOT NULL, + table_name TEXT NOT NULL, + action TEXT NOT NULL, + previous_state JSONB, + new_state JSONB, + actor TEXT NOT NULL, + approved_by TEXT, + reason TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +ALTER TABLE meta.ingestion_ownership_audit ADD COLUMN IF NOT EXISTS db_user TEXT NOT NULL DEFAULT session_user; + +CREATE INDEX IF NOT EXISTS ix_ownership_audit_source ON meta.ingestion_ownership_audit (source_id, created_at); + +CREATE OR REPLACE FUNCTION meta.protect_ownership_audit() +RETURNS trigger +LANGUAGE plpgsql +AS $$ +BEGIN + IF COALESCE(current_setting('fbi.audit_maintenance', TRUE), '') = 'on' THEN + RETURN CASE TG_OP WHEN 'DELETE' THEN OLD ELSE NEW END; + END IF; + RAISE EXCEPTION 'meta.ingestion_ownership_audit is append-only (% rejected)', TG_OP USING ERRCODE = 'FBI03'; +END; +$$; + +DROP TRIGGER IF EXISTS trg_ownership_audit_append_only ON meta.ingestion_ownership_audit; +CREATE TRIGGER trg_ownership_audit_append_only + BEFORE UPDATE OR DELETE ON meta.ingestion_ownership_audit + FOR EACH ROW EXECUTE FUNCTION meta.protect_ownership_audit(); + +CREATE TABLE IF NOT EXISTS meta.cutover_approval ( + approval_id BIGSERIAL PRIMARY KEY, + source_id TEXT NOT NULL, + table_name TEXT NOT NULL, + epoch BIGINT NOT NULL, + approved_db_user TEXT NOT NULL DEFAULT session_user, + approved_label TEXT, + approved_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + UNIQUE (source_id, table_name, epoch, approved_db_user) +); + +CREATE OR REPLACE FUNCTION meta.approve_cutover(p_source_id TEXT, p_table_name TEXT, p_label TEXT DEFAULT NULL) +RETURNS BIGINT +LANGUAGE plpgsql +AS $$ +DECLARE + o meta.ingestion_ownership%ROWTYPE; + v_id BIGINT; +BEGIN + SELECT * INTO o FROM meta.ingestion_ownership WHERE source_id = p_source_id AND table_name = p_table_name; + IF NOT FOUND OR o.cutover_state <> 'CUTOVER_READY' THEN + RAISE EXCEPTION '%/% is not CUTOVER_READY', p_source_id, p_table_name USING ERRCODE = 'FBI03'; + END IF; + IF o.shadow_requested_db_user IS NOT DISTINCT FROM session_user THEN + RAISE EXCEPTION 'The database login that requested the shadow (%) cannot approve its own cutover', session_user + USING ERRCODE = 'FBI03'; + END IF; + INSERT INTO meta.cutover_approval (source_id, table_name, epoch, approved_label) + VALUES (p_source_id, p_table_name, o.epoch, p_label) + ON CONFLICT (source_id, table_name, epoch, approved_db_user) DO UPDATE SET approved_at = NOW() + RETURNING approval_id INTO v_id; + RETURN v_id; +END; +$$; + +CREATE OR REPLACE FUNCTION meta.guard_ownership_update() +RETURNS trigger +LANGUAGE plpgsql +AS $$ +DECLARE + v_max_age INTEGER := meta.setting('reconciliation_max_age_minutes', '120')::INTEGER; + v_mode TEXT := meta.setting('dual_control_mode', 'parameter'); + v_ok BOOLEAN; +BEGIN + IF NEW.engine IS DISTINCT FROM OLD.engine THEN + IF OLD.engine = 'NONE' OR NEW.engine = 'NONE' THEN + NULL; + ELSIF OLD.shadow_engine IS NOT NULL AND NEW.engine = OLD.shadow_engine + AND NEW.shadow_engine IS NULL AND OLD.cutover_state = 'CUTOVER_READY' THEN + SELECT EXISTS ( + SELECT 1 FROM meta.reconciliation_run r + WHERE r.reconciliation_id = OLD.last_reconciliation_id + AND r.source_id = OLD.source_id + AND r.table_name = OLD.table_name + AND r.comparison = 'shadow_vs_prod' + AND r.status = 'PASSED' + AND r.finished_at > NOW() - make_interval(mins => v_max_age) + ) INTO v_ok; + IF NOT v_ok THEN + RAISE EXCEPTION 'Cutover of %/% needs a PASSED shadow_vs_prod reconciliation of this table newer than % minutes', + OLD.source_id, OLD.table_name, v_max_age USING ERRCODE = 'FBI03'; + END IF; + IF v_mode = 'database' THEN + SELECT EXISTS ( + SELECT 1 FROM meta.cutover_approval a + WHERE a.source_id = OLD.source_id AND a.table_name = OLD.table_name AND a.epoch = OLD.epoch + AND a.approved_db_user <> COALESCE(OLD.shadow_requested_db_user, '') + AND a.approved_db_user <> session_user + ) INTO v_ok; + IF NOT v_ok THEN + RAISE EXCEPTION 'Cutover of %/% needs an approval recorded by a different database login (meta.approve_cutover)', + OLD.source_id, OLD.table_name USING ERRCODE = 'FBI03'; + END IF; + END IF; + ELSIF COALESCE(current_setting('fbi.force_approved_by', TRUE), '') <> '' THEN + NULL; + ELSE + RAISE EXCEPTION 'Illegal engine change % -> % for %/%', OLD.engine, NEW.engine, OLD.source_id, OLD.table_name + USING ERRCODE = 'FBI03'; + END IF; + END IF; + IF (NEW.engine IS DISTINCT FROM OLD.engine OR NEW.shadow_engine IS DISTINCT FROM OLD.shadow_engine) + AND NEW.epoch <= OLD.epoch THEN + RAISE EXCEPTION 'Ownership of %/% changed without a higher epoch', OLD.source_id, OLD.table_name + USING ERRCODE = 'FBI03'; + END IF; + IF NEW.shadow_engine IS NOT NULL AND OLD.shadow_engine IS NULL THEN + NEW.shadow_requested_db_user := session_user; + ELSIF NEW.shadow_engine IS NULL THEN + NEW.shadow_requested_db_user := NULL; + NEW.shadow_requested_by := NULL; + END IF; + RETURN NEW; +END; +$$; + +DROP TRIGGER IF EXISTS trg_ownership_guard ON meta.ingestion_ownership; +CREATE TRIGGER trg_ownership_guard + BEFORE UPDATE ON meta.ingestion_ownership + FOR EACH ROW EXECUTE FUNCTION meta.guard_ownership_update(); + +CREATE OR REPLACE FUNCTION meta.ownership_lock_key(p_source_id TEXT, p_table_name TEXT) +RETURNS BIGINT +LANGUAGE sql +STABLE +AS $$ + SELECT hashtextextended(p_table_name || chr(31) || p_source_id, 7) +$$; + +CREATE OR REPLACE FUNCTION meta.lock_ownership_exclusive(p_source_id TEXT, p_table_name TEXT) +RETURNS VOID +LANGUAGE sql +AS $$ + SELECT pg_advisory_xact_lock(meta.ownership_lock_key(p_source_id, p_table_name)) +$$; + +CREATE TABLE IF NOT EXISTS meta.consumer_offsets ( + consumer_group TEXT NOT NULL, + topic TEXT NOT NULL, + kafka_partition INTEGER NOT NULL, + next_offset BIGINT NOT NULL, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (consumer_group, topic, kafka_partition) +); + +CREATE TABLE IF NOT EXISTS meta.extract_watermarks ( + source_id TEXT NOT NULL, + table_name TEXT NOT NULL, + target TEXT NOT NULL CHECK (target IN ('prod', 'shadow')), + cursor_value TIMESTAMPTZ, + last_pk BIGINT, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (source_id, table_name, target) +); + +CREATE TABLE IF NOT EXISTS meta.ingestion_progress ( + source_id TEXT NOT NULL, + table_name TEXT NOT NULL, + target TEXT NOT NULL, + writer_id TEXT NOT NULL, + engine TEXT NOT NULL, + rows_applied BIGINT NOT NULL DEFAULT 0, + last_source_ts TIMESTAMPTZ, + last_source_lsn BIGINT, + last_applied_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + PRIMARY KEY (source_id, table_name, target, writer_id) +); + +CREATE TABLE IF NOT EXISTS meta.ingestion_audit ( + table_name TEXT NOT NULL, + writer_id TEXT NOT NULL, + minute_bucket TIMESTAMPTZ NOT NULL, + events_consumed BIGINT NOT NULL DEFAULT 0, + rows_applied BIGINT NOT NULL DEFAULT 0, + rows_skipped BIGINT NOT NULL DEFAULT 0, + events_dead_lettered BIGINT NOT NULL DEFAULT 0, + PRIMARY KEY (table_name, writer_id, minute_bucket) +); + +DO $$ +BEGIN + IF to_regclass('meta.entity_change_log') IS NOT NULL + AND (SELECT relkind FROM pg_class WHERE oid = to_regclass('meta.entity_change_log')) = 'r' THEN + ALTER TABLE meta.entity_change_log RENAME TO entity_change_log_legacy; + ALTER TABLE meta.entity_change_log_legacy RENAME CONSTRAINT entity_change_log_pkey TO entity_change_log_legacy_pkey; + ALTER SEQUENCE meta.entity_change_log_change_id_seq OWNED BY NONE; + END IF; +END +$$; + +CREATE TABLE IF NOT EXISTS meta.entity_change_log ( + change_id BIGINT NOT NULL DEFAULT 0, + source_id TEXT NOT NULL, + tenant_id TEXT NOT NULL, + entity TEXT NOT NULL CHECK (entity IN ('loan', 'client', 'global', 'source')), + entity_id BIGINT NOT NULL, + min_affected_date DATE NOT NULL, + changed_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + txid xid8 NOT NULL DEFAULT pg_current_xact_id(), + PRIMARY KEY (changed_at, change_id) +) PARTITION BY RANGE (changed_at); + +CREATE SEQUENCE IF NOT EXISTS meta.entity_change_log_change_id_seq; +ALTER TABLE meta.entity_change_log ALTER COLUMN change_id SET DEFAULT nextval('meta.entity_change_log_change_id_seq'); + +CREATE TABLE IF NOT EXISTS meta.entity_change_log_default PARTITION OF meta.entity_change_log DEFAULT; + +CREATE INDEX IF NOT EXISTS ix_entity_change_log_txid ON meta.entity_change_log (txid); +CREATE INDEX IF NOT EXISTS ix_entity_change_log_source ON meta.entity_change_log (source_id, entity, entity_id); + +CREATE OR REPLACE FUNCTION meta.ensure_change_log_partitions(p_days_back INTEGER DEFAULT 1, p_days_ahead INTEGER DEFAULT 3) +RETURNS INTEGER +LANGUAGE plpgsql +SECURITY DEFINER +SET search_path = meta, pg_temp +AS $$ +DECLARE + d DATE; + v_name TEXT; + n INTEGER := 0; +BEGIN + FOR d IN SELECT generate_series(CURRENT_DATE - p_days_back, CURRENT_DATE + p_days_ahead, INTERVAL '1 day')::DATE LOOP + v_name := 'entity_change_log_' || to_char(d, 'YYYYMMDD'); + IF to_regclass('meta.' || v_name) IS NULL THEN + IF EXISTS ( + SELECT 1 FROM meta.entity_change_log_default + WHERE changed_at >= d::TIMESTAMPTZ AND changed_at < (d + 1)::TIMESTAMPTZ + ) THEN + EXECUTE format('CREATE TABLE meta.%I (LIKE meta.entity_change_log INCLUDING DEFAULTS INCLUDING CONSTRAINTS)', v_name); + EXECUTE format( + 'WITH moved AS (DELETE FROM meta.entity_change_log_default WHERE changed_at >= %L AND changed_at < %L ' + 'RETURNING *) INSERT INTO meta.%I SELECT * FROM moved', + d::TIMESTAMPTZ, (d + 1)::TIMESTAMPTZ, v_name); + EXECUTE format( + 'ALTER TABLE meta.entity_change_log ATTACH PARTITION meta.%I FOR VALUES FROM (%L) TO (%L)', + v_name, d::TIMESTAMPTZ, (d + 1)::TIMESTAMPTZ); + ELSE + EXECUTE format( + 'CREATE TABLE meta.%I PARTITION OF meta.entity_change_log FOR VALUES FROM (%L) TO (%L)', + v_name, d::TIMESTAMPTZ, (d + 1)::TIMESTAMPTZ); + END IF; + n := n + 1; + END IF; + END LOOP; + RETURN n; +END; +$$; + +DO $$ +BEGIN + IF to_regclass('meta.entity_change_log_legacy') IS NOT NULL THEN + INSERT INTO meta.entity_change_log (change_id, source_id, tenant_id, entity, entity_id, min_affected_date, changed_at, txid) + SELECT change_id, source_id, tenant_id, entity, entity_id, min_affected_date, changed_at, '0'::xid8 + FROM meta.entity_change_log_legacy; + PERFORM setval('meta.entity_change_log_change_id_seq', + GREATEST((SELECT COALESCE(MAX(change_id), 0) FROM meta.entity_change_log_legacy), 1)); + DROP TABLE meta.entity_change_log_legacy; + END IF; +END +$$; + +SELECT meta.ensure_change_log_partitions(); + +CREATE TABLE IF NOT EXISTS meta.dbt_change_cursor ( + model_name TEXT PRIMARY KEY, + last_change_id BIGINT NOT NULL DEFAULT 0, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); +ALTER TABLE meta.dbt_change_cursor ADD COLUMN IF NOT EXISTS last_txid xid8 NOT NULL DEFAULT '0'::xid8; +ALTER TABLE meta.dbt_change_cursor ADD COLUMN IF NOT EXISTS last_run_id BIGINT NOT NULL DEFAULT 0; + +CREATE SEQUENCE IF NOT EXISTS meta.dbt_run_seq; + +CREATE TABLE IF NOT EXISTS meta.dbt_mart_cursor ( + model_name TEXT PRIMARY KEY, + last_run_id BIGINT NOT NULL DEFAULT 0, + updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); + +CREATE TABLE IF NOT EXISTS meta.dbt_mart_dirty ( + run_id BIGINT NOT NULL, + source_id TEXT NOT NULL, + office_id BIGINT, + product_id BIGINT, + from_date DATE NOT NULL, + changed_from DATE, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW() +); +CREATE INDEX IF NOT EXISTS ix_dbt_mart_dirty_run ON meta.dbt_mart_dirty (run_id); + +CREATE OR REPLACE FUNCTION meta.change_log_horizon() +RETURNS xid8 +LANGUAGE sql +VOLATILE +AS $$ + SELECT pg_snapshot_xmin(pg_current_snapshot()) +$$; + +CREATE OR REPLACE FUNCTION meta.purge_change_log(p_retention_days INTEGER DEFAULT NULL) +RETURNS INTEGER +LANGUAGE plpgsql +SECURITY DEFINER +SET search_path = meta, pg_temp +AS $$ +DECLARE + v_days INTEGER := COALESCE(p_retention_days, meta.setting('change_log_retention_days', '14')::INTEGER); + v_stale INTEGER := meta.setting('cursor_stale_days', '7')::INTEGER; + v_floor xid8; + r RECORD; + v_bound DATE; + v_has_pending BOOLEAN; + n INTEGER := 0; +BEGIN + SELECT MIN(last_txid) INTO v_floor + FROM meta.dbt_change_cursor + WHERE updated_at > NOW() - make_interval(days => v_stale); + IF v_floor IS NULL THEN + RETURN 0; + END IF; + PERFORM meta.ensure_change_log_partitions(); + FOR r IN + SELECT c.relname + FROM pg_inherits i + JOIN pg_class c ON c.oid = i.inhrelid + WHERE i.inhparent = 'meta.entity_change_log'::regclass + AND c.relname ~ '^entity_change_log_[0-9]{8}$' + LOOP + v_bound := to_date(right(r.relname, 8), 'YYYYMMDD'); + IF v_bound + 1 > (NOW() - make_interval(days => v_days))::DATE THEN + CONTINUE; + END IF; + EXECUTE format('SELECT EXISTS (SELECT 1 FROM meta.%I WHERE txid >= $1)', r.relname) INTO v_has_pending USING v_floor; + IF NOT v_has_pending THEN + EXECUTE format('DROP TABLE meta.%I', r.relname); + n := n + 1; + END IF; + END LOOP; + DELETE FROM meta.entity_change_log_default + WHERE changed_at < NOW() - make_interval(days => v_days) AND txid < v_floor; + RETURN n; +END; +$$; + +CREATE OR REPLACE FUNCTION meta.purge_progress(p_days INTEGER DEFAULT NULL) +RETURNS BIGINT +LANGUAGE plpgsql +AS $$ +DECLARE + v_days INTEGER := COALESCE(p_days, meta.setting('progress_retention_days', '7')::INTEGER); + v_a BIGINT; + v_b BIGINT; +BEGIN + DELETE FROM meta.ingestion_progress WHERE last_applied_at < NOW() - make_interval(days => v_days); + GET DIAGNOSTICS v_a = ROW_COUNT; + DELETE FROM meta.ingestion_audit WHERE minute_bucket < NOW() - make_interval(days => v_days); + GET DIAGNOSTICS v_b = ROW_COUNT; + RETURN v_a + v_b; +END; +$$; + +CREATE OR REPLACE FUNCTION meta.purge_ownership_audit(p_days INTEGER DEFAULT NULL) Review Comment: Fixed. All SECURITY DEFINER functions are revoked from PUBLIC and granted only to a fbi_control_plane group role, which the loader is a member of. Their arguments are clamped to the retention settings in meta.platform_settings, which only the owner can change. So purge_ownership_audit(0) can't go below the policy, and ensure_change_log_partitions is capped at 62 days back and 31 ahead. I added a test showing the reader and engine roles get 42501 on all four functions. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
