Oracle to PostgreSQL Migration: Comprehensive Guide Using AWS DMS & Ora2Pg

Pre-Migration Assessment:
├── Database Size Analysis
│   ├── Total Data Volume
│   ├── Table Sizes
│   └── Growth Projections
├── Schema Complexity
│   ├── Number of Objects
│   ├── Custom Data Types
│   └── Stored Procedures
├── Dependencies
│   ├── Application Code
│   ├── External Systems
│   └── Reporting Tools
└── Performance Requirements
    ├── Throughput
    ├── Latency
    └── Concurrency

AWS DMS Advantages:
  - Minimal downtime
  - Continuous replication
  - AWS integration
  - Managed service
  - Real-time monitoring

Ora2Pg Advantages:
  - Detailed schema conversion
  - PL/SQL to PL/pgSQL conversion
  - Open-source
  - One-time migration
  - Local control

{
  "ReplicationInstance": {
    "ReplicationInstanceClass": "dms.r5.2xlarge",
    "AllocatedStorage": 100,
    "MultiAZ": true,
    "PubliclyAccessible": false,
    "VpcSecurityGroups": ["sg-xxxxx"]
  },
  "Source": {
    "EngineName": "oracle",
    "ServerName": "oracle-source.xxxx.xx",
    "Port": 1521,
    "DatabaseName": "PRODDB",
    "Username": "dms_user",
    "Password": "xxx"
  },
  "Target": {
    "EngineName": "postgres",
    "ServerName": "postgresql-target.xxxx.xx",
    "Port": 5432,
    "DatabaseName": "proddb",
    "Username": "dms_user",
    "Password": "xxx"
  }
}

{
  "Rules": [
    {
      "RuleType": "selection",
      "RuleId": "1",
      "RuleAction": "include",
      "ObjectLocator": {
        "SchemaName": "PROD",
        "TableName": "%"
      }
    },
    {
      "RuleType": "transformation",
      "RuleId": "2",
      "RuleAction": "convert-lowercase",
      "RuleTarget": "schema",
      "ObjectLocator": {
        "SchemaName": "PROD"
      }
    }
  ]
}

CloudWatch Metrics:
  - FullLoadThroughputRows
  - CDCLatency
  - CPUUtilization
  - FreeStorageSpace

Alarms:
  CDCLatencyWarning:
    Threshold: 300  # seconds
    Period: 300
    EvaluationPeriods: 2
    
  ReplicationLag:
    Threshold: 1800  # seconds
    Period: 300
    EvaluationPeriods: 3

# ora2pg.conf
ORACLE_HOME     /u01/app/oracle/product/19.0.0/dbhome_1
ORACLE_DSN      dbi:Oracle:host=oracle-server;sid=PRODDB;port=1521
ORACLE_USER     migration_user
ORACLE_PWD      secret_password
SCHEMA          PROD

POSTGRES_DSN    dbi:Pg:dbname=proddb;host=postgresql-server;port=5432
POSTGRES_USER   postgres
POSTGRES_PWD    secret_password

EXPORT_SCHEMA   1
EXPORT_TABLES   1
EXPORT_TRIGGERS 1
PARALLEL        4
TYPE_FUNCTION   1

# Export schema
ora2pg -c config/ora2pg.conf -t SCHEMA -o schema.sql

# Export functions and procedures
ora2pg -c config/ora2pg.conf -t PACKAGE -o functions.sql

# Export triggers
ora2pg -c config/ora2pg.conf -t TRIGGER -o triggers.sql

#!/bin/bash
# migrate_data.sh

TABLES=$(ora2pg -c config/ora2pg.conf -t SHOW_TABLES)

for TABLE in $TABLES; do
    echo "Migrating $TABLE..."
    ora2pg -c config/ora2pg.conf -t COPY -a -n $TABLE \
        --nls_lang "AMERICAN_AMERICA.AL32UTF8" \
        --parallel 4 \
        --batch_size 10000
done

-- Oracle to PostgreSQL Type Mapping
CREATE OR REPLACE FUNCTION convert_oracle_type(
    oracle_type text
) RETURNS text AS $$
BEGIN
    RETURN CASE oracle_type
        WHEN 'NUMBER' THEN 'numeric'
        WHEN 'VARCHAR2' THEN 'varchar'
        WHEN 'DATE' THEN 'timestamp'
        WHEN 'CLOB' THEN 'text'
        WHEN 'BLOB' THEN 'bytea'
        ELSE oracle_type
    END;
END;
$$ LANGUAGE plpgsql;

-- Oracle Package Header
CREATE OR REPLACE PACKAGE employee_mgmt AS
    PROCEDURE add_employee(
        p_name IN VARCHAR2,
        p_salary IN NUMBER
    );
    FUNCTION get_salary(p_emp_id IN NUMBER) RETURN NUMBER;
END employee_mgmt;

-- PostgreSQL Equivalent
CREATE SCHEMA employee_mgmt;

CREATE OR REPLACE FUNCTION employee_mgmt.add_employee(
    p_name varchar,
    p_salary numeric
) RETURNS void AS $$
BEGIN
    INSERT INTO employees (name, salary)
    VALUES (p_name, p_salary);
END;
$$ LANGUAGE plpgsql;

CREATE OR REPLACE FUNCTION employee_mgmt.get_salary(
    p_emp_id numeric
) RETURNS numeric AS $$
BEGIN
    RETURN salary FROM employees WHERE emp_id = p_emp_id;
END;
$$ LANGUAGE plpgsql;

{
  "TaskSettings": {
    "TargetMetadata": {
      "BatchApplyEnabled": true,
      "BatchApplyPreserveTransaction": true,
      "BatchSize": 50000,
      "CommitRate": 10000
    },
    "FullLoadSettings": {
      "MaxFullLoadSubTasks": 8,
      "TransactionConsistencyTimeout": 600,
      "CommitRate": 10000
    },
    "ChangeProcessingTuning": {
      "BatchApplyPreserveTransaction": true,
      "BatchApplyTimeoutMin": 1,
      "BatchApplyTimeoutMax": 30,
      "BatchSize": 10000,
      "CommitTimeout": 1
    }
  }
}

# parallel_load.py
import multiprocessing
import psycopg2
from concurrent.futures import ThreadPoolExecutor

def load_table_chunk(params):
    table, start_id, end_id = params
    
    source_conn = psycopg2.connect("oracle connection string")
    target_conn = psycopg2.connect("postgresql connection string")
    
    try:
        with source_conn.cursor() as src_cur, target_conn.cursor() as tgt_cur:
            src_cur.execute(f"""
                SELECT * FROM {table}
                WHERE id BETWEEN {start_id} AND {end_id}
            """)
            
            data = src_cur.fetchall()
            tgt_cur.executemany(f"""
                INSERT INTO {table} VALUES %s
            """, data)
            
        target_conn.commit()
    finally:
        source_conn.close()
        target_conn.close()

def parallel_table_load(table, total_rows, chunk_size=100000):
    chunks = [(table, i, i + chunk_size - 1) 
             for i in range(1, total_rows, chunk_size)]
    
    with ThreadPoolExecutor(max_workers=8) as executor:
        executor.map(load_table_chunk, chunks)

-- Compare Row Counts
CREATE OR REPLACE FUNCTION validate_table_counts()
RETURNS TABLE (
    table_name text,
    oracle_count bigint,
    pg_count bigint,
    is_match boolean
) AS $$
DECLARE
    tables CURSOR FOR 
        SELECT tablename FROM pg_tables 
        WHERE schemaname = 'public';
BEGIN
    FOR table_record IN tables LOOP
        RETURN QUERY
        SELECT 
            table_record.tablename,
            oracle_count.cnt::bigint,
            pg_count.cnt::bigint,
            oracle_count.cnt = pg_count.cnt
        FROM 
            dblink('oracle_connection',
                format('SELECT COUNT(*) as cnt FROM %I', 
                    table_record.tablename)
            ) AS oracle_count(cnt bigint)
            CROSS JOIN
            (SELECT COUNT(*) as cnt 
             FROM public.table_record.tablename) pg_count;
    END LOOP;
END;
$$ LANGUAGE plpgsql;

-- Analyze Tables and Create Missing Indexes
CREATE OR REPLACE FUNCTION optimize_indexes()
RETURNS void AS $$
DECLARE
    table_record record;
BEGIN
    -- Analyze all tables
    FOR table_record IN 
        SELECT tablename FROM pg_tables 
        WHERE schemaname = 'public'
    LOOP
        EXECUTE format('ANALYZE %I', table_record.tablename);
    END LOOP;
    
    -- Create missing indexes based on query patterns
    FOR table_record IN 
        SELECT schemaname, tablename 
        FROM pg_tables 
        WHERE schemaname = 'public'
    LOOP
        EXECUTE format('
            SELECT create_missing_indexes(
                %L, %L, 
                (SELECT COUNT(*) FROM %I.%I)
            )',
            table_record.schemaname,
            table_record.tablename,
            table_record.schemaname,
            table_record.tablename
        );
    END LOOP;
END;
$$ LANGUAGE plpgsql;

# postgresql.conf optimizations
max_connections = 200
shared_buffers = 8GB
effective_cache_size = 24GB
maintenance_work_mem = 2GB
checkpoint_completion_target = 0.9
wal_buffers = 16MB
default_statistics_target = 100
random_page_cost = 1.1
effective_io_concurrency = 200
work_mem = 52428kB
min_wal_size = 1GB
max_wal_size = 4GB
max_worker_processes = 8
max_parallel_workers_per_gather = 4
max_parallel_workers = 8

-- Convert Large Tables to Partitioned Tables
CREATE TABLE orders_new (
    order_id bigint,
    order_date timestamp,
    customer_id bigint,
    total_amount numeric,
    status varchar(20)
) PARTITION BY RANGE (order_date);

-- Create Monthly Partitions
CREATE TABLE orders_2025_01 PARTITION OF orders_new
    FOR VALUES FROM ('2025-01-01') TO ('2025-02-01');
    
CREATE TABLE orders_2025_02 PARTITION OF orders_new
    FOR VALUES FROM ('2025-02-01') TO ('2025-03-01');

-- Create Indexes on Partitions
CREATE INDEX idx_orders_2025_01_customer 
    ON orders_2025_01 (customer_id);
CREATE INDEX idx_orders_2025_02_customer 
    ON orders_2025_02 (customer_id);

-- Check for Invalid Objects
SELECT schemaname, tablename, tableowner 
FROM pg_tables 
WHERE tablename IN (
    SELECT relname 
    FROM pg_class 
    WHERE relkind = 'r' 
    AND NOT EXISTS (
        SELECT 1 
        FROM pg_attribute 
        WHERE attrelid = pg_class.oid 
        AND NOT attisdropped
    )
);

-- Identify Locked Objects
SELECT blocked_locks.pid AS blocked_pid,
       blocked_activity.usename AS blocked_user,
       blocking_locks.pid AS blocking_pid,
       blocking_activity.usename AS blocking_user,
       blocked_activity.query AS blocked_statement
FROM pg_catalog.pg_locks blocked_locks
JOIN pg_catalog.pg_stat_activity blocked_activity 
    ON blocked_activity.pid = blocked_locks.pid
JOIN pg_catalog.pg_locks blocking_locks 
    ON blocking_locks.locktype = blocked_locks.locktype
JOIN pg_catalog.pg_stat_activity blocking_activity 
    ON blocking_activity.pid = blocking_locks.pid
WHERE NOT blocked_locks.granted;

#!/bin/bash
# error_handler.sh

log_error() {
    local timestamp=$(date '+%Y-%m-%d %H:%M:%S')
    local error_msg="$1"
    local error_file="migration_errors.log"
    
    echo "[$timestamp] ERROR: $error_msg" >> "$error_file"
    
    # Send notification if critical
    if [[ "$error_msg" == *"CRITICAL"* ]]; then
        send_notification "$error_msg"
    fi
}

send_notification() {
    local message="$1"
    # Add your notification logic (email, Slack, etc.)
    curl -X POST -H 'Content-type: application/json' \
        --data "{\"text\":\"$message\"}" \
        "$WEBHOOK_URL"
}