airajena commented on code in PR #102: URL: https://github.com/apache/fineract-business-intelligence/pull/102#discussion_r4318432587
########## warehouse/schema/platform.sql: ########## @@ -0,0 +1,733 @@ +-- 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, + +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.'), + ('ownership_lock_buckets', '64', TRUE, + 'Advisory-lock buckets per table used to fence writers against ownership transitions.'), + ('require_dual_control', 'true', FALSE, + 'When true a cutover must be approved by a different actor than the requester.'), + ('change_log_retention_days', '14', FALSE, + 'Consumed entity_change_log rows older than this are purged by maintenance.') +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() +); + +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 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, 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, + (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, + 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|vault):') +); + +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')) + ) +); + +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() +); + +CREATE INDEX IF NOT EXISTS ix_ownership_audit_source ON meta.ingestion_ownership_audit (source_id, created_at); + +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 || ':' || ( + abs(hashtextextended(p_source_id, 0)) % meta.setting('ownership_lock_buckets', '64')::BIGINT + )::TEXT, + 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) +); + +CREATE TABLE IF NOT EXISTS meta.entity_change_log ( Review Comment: Fixed. entity_change_log is range-partitioned by day and retention drops whole partitions. Stale cursors no longer pin the purge. I also added retention for ingestion_progress and the ownership audit. -- 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]
