pg_fact_loader

pig build pkg pg_fact_loader         # build DEB packages

pig repo add pgdg -u          # Add PGDG repo and update cache

CREATE EXTENSION pg_fact_loader;

CREATE EXTENSION pg_fact_loader;

CREATE EXTENSION pglogical;
CREATE EXTENSION pglogical_ticker;
CREATE EXTENSION pg_fact_loader;

-- Register a fact table
INSERT INTO fact_loader.fact_tables (fact_table_relid, fact_table_agg_proid, ...)
VALUES ('public.customers_fact'::regclass, 'customers_fact_merge'::regproc, ...);

-- Register queue (audit) tables
INSERT INTO fact_loader.queue_tables (queue_table_relid, queue_of_base_table_relid, ...)
VALUES ('audit.customers_audit'::regclass, 'public.customers'::regclass, ...);

-- Connect queue tables to fact tables with merge functions
INSERT INTO fact_loader.queue_table_deps
    (fact_table_id, queue_table_id, insert_merge_proid, update_merge_proid, delete_merge_proid)
VALUES (1, 1, 'customers_fact_merge'::regproc, 'customers_fact_merge'::regproc, 'customers_fact_merge'::regproc);

-- Define how to retrieve the key from queue entries
INSERT INTO fact_loader.key_retrieval_sequences
    (queue_table_dep_id, return_columns, is_fact_key)
VALUES (1, '{customer_id}', true);

-- Process pending changes
SELECT fact_loader.worker();

-- Schedule this to run periodically (e.g., every few seconds via pg_cron)

-- Run the merge function for every existing row
SELECT customers_fact_merge(customer_id) FROM customers;

SELECT fact_loader.add_batch_id_fields();