Set up Oracle LogMiner CDC end to end

-- Oracle, as SYSDBA, in the PDB
ALTER SESSION SET CONTAINER = ORCL;

CREATE USER SHOP IDENTIFIED BY "<shop password>" DEFAULT TABLESPACE USERS QUOTA UNLIMITED ON USERS;
GRANT CREATE SESSION TO SHOP;

CREATE TABLE SHOP.CUSTOMERS (
  ID           NUMBER(10)    PRIMARY KEY,
  FULL_NAME    VARCHAR2(100) NOT NULL,
  EMAIL        VARCHAR2(150),
  COUNTRY      VARCHAR2(2),
  TIER         VARCHAR2(10),
  STATUS       VARCHAR2(10),
  CREDIT_LIMIT NUMBER(12,2),
  CREATED_AT   TIMESTAMP DEFAULT SYSTIMESTAMP
);

CREATE TABLE SHOP.ORDERS (
  ID          NUMBER(10)    PRIMARY KEY,
  CUSTOMER_ID NUMBER(10)    NOT NULL,
  AMOUNT      NUMBER(12,2),
  CURRENCY    VARCHAR2(3),
  STATUS      VARCHAR2(10),
  ORDER_DATE  DATE
);

INSERT INTO SHOP.CUSTOMERS
SELECT LEVEL, 'Customer '||LEVEL, 'customer'||LEVEL||'@example.com',
  CASE MOD(LEVEL,8) WHEN 0 THEN 'IT' WHEN 1 THEN 'DE' WHEN 2 THEN 'US' WHEN 3 THEN 'FR'
                    WHEN 4 THEN 'ES' WHEN 5 THEN 'GB' WHEN 6 THEN 'JP' ELSE 'BR' END,
  CASE MOD(LEVEL,4) WHEN 0 THEN 'bronze' WHEN 1 THEN 'silver' WHEN 2 THEN 'gold' ELSE 'platinum' END,
  CASE WHEN MOD(LEVEL,7)=0 THEN 'suspended' WHEN MOD(LEVEL,5)=0 THEN 'inactive' ELSE 'active' END,
  1000 + MOD(LEVEL,10)*500, SYSTIMESTAMP
FROM DUAL CONNECT BY LEVEL <= 200;

INSERT INTO SHOP.ORDERS
SELECT LEVEL, MOD(LEVEL,200)+1, ROUND(10 + MOD(LEVEL*37,490) + MOD(LEVEL,100)/100,2),
  CASE MOD(LEVEL,3) WHEN 0 THEN 'EUR' WHEN 1 THEN 'USD' ELSE 'GBP' END,
  CASE MOD(LEVEL,5) WHEN 0 THEN 'new' WHEN 1 THEN 'paid' WHEN 2 THEN 'shipped'
                    WHEN 3 THEN 'delivered' ELSE 'cancelled' END,
  TRUNC(SYSDATE) - MOD(LEVEL,90)
FROM DUAL CONNECT BY LEVEL <= 1000;
COMMIT;

-- PostgreSQL target
CREATE SCHEMA IF NOT EXISTS ora_cdc;

CREATE TABLE ora_cdc.customers (
  id           numeric(10,0) PRIMARY KEY,
  full_name    varchar(100)  NOT NULL,
  email        varchar(150),
  country      varchar(2),
  tier         varchar(10),
  status       varchar(10),
  credit_limit numeric(12,2),
  created_at   timestamp
);

CREATE TABLE ora_cdc.orders (
  id          numeric(10,0) PRIMARY KEY,
  customer_id numeric(10,0) NOT NULL,
  amount      numeric(12,2),
  currency    varchar(3),
  status      varchar(10),
  order_date  timestamp
);

SELECT log_mode FROM v$database;

SHUTDOWN IMMEDIATE
STARTUP MOUNT
ALTER DATABASE ARCHIVELOG;
ALTER DATABASE OPEN;

ALTER PLUGGABLE DATABASE ALL OPEN;
ALTER PLUGGABLE DATABASE ALL SAVE STATE;

ARCHIVE LOG LIST

ALTER DATABASE ADD SUPPLEMENTAL LOG DATA;
ALTER DATABASE ADD SUPPLEMENTAL LOG DATA (PRIMARY KEY) COLUMNS;

SELECT supplemental_log_data_min, supplemental_log_data_pk FROM v$database;

CREATE USER C##GS_LOGMINER IDENTIFIED BY "<LogMiner password>" CONTAINER=ALL;

GRANT CREATE SESSION,
      LOGMINING,
      SELECT ANY TRANSACTION,
      EXECUTE_CATALOG_ROLE,
      SELECT_CATALOG_ROLE
   TO C##GS_LOGMINER CONTAINER=ALL;

ALTER SESSION SET CONTAINER = ORCL;

CREATE USER GS_READER IDENTIFIED BY "<data user password>";
GRANT CREATE SESSION TO GS_READER;

ALTER TABLE SHOP.CUSTOMERS ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;
ALTER TABLE SHOP.ORDERS    ADD SUPPLEMENTAL LOG DATA (ALL) COLUMNS;

GRANT SELECT ON SHOP.CUSTOMERS TO GS_READER;
GRANT SELECT ON SHOP.ORDERS    TO GS_READER;
GRANT SELECT ON SHOP.CUSTOMERS TO C##GS_LOGMINER;
GRANT SELECT ON SHOP.ORDERS    TO C##GS_LOGMINER;

SELECT 'customers' AS table_name, count(*) FROM ora_cdc.customers
UNION ALL
SELECT 'orders', count(*) FROM ora_cdc.orders;

-- Oracle
INSERT INTO SHOP.CUSTOMERS (ID, FULL_NAME, EMAIL, COUNTRY, TIER, STATUS, CREDIT_LIMIT)
  VALUES (201, 'Ada Lovelace', 'ada@example.com', 'GB', 'gold', 'active', 5000);
COMMIT;

-- PostgreSQL, after about 10 seconds
SELECT id, full_name, tier, credit_limit FROM ora_cdc.customers WHERE id = 201;

-- Oracle
UPDATE SHOP.CUSTOMERS SET TIER = 'platinum', CREDIT_LIMIT = 9000 WHERE ID = 201;
COMMIT;

-- PostgreSQL
SELECT id, full_name, tier, credit_limit FROM ora_cdc.customers WHERE id = 201;

-- Oracle
DELETE FROM SHOP.CUSTOMERS WHERE ID = 201;
COMMIT;

-- PostgreSQL
SELECT count(*) FROM ora_cdc.customers WHERE id = 201;