Migrating from Oracle to PostgreSQL: A Comprehensive Guide

class OracleAssessmentTool:
    def __init__(self):
        self.complexity_factors = {
            'schema_complexity': {
                'table_count': {'low': 50, 'medium': 200, 'high': 500},
                'index_count': {'low': 100, 'medium': 500, 'high': 1000},
                'constraint_count': {'low': 100, 'medium': 300, 'high': 600}
            },
            'feature_usage': {
                'oracle_specific_features': [
                    'partitioning', 'materialized_views', 'packages', 
                    'triggers', 'functions', 'sequences', 'synonyms'
                ],
                'advanced_features': [
                    'xml_data_types', 'spatial_data', 'advanced_security',
                    'advanced_analytics', 'data_mining'
                ]
            },
            'data_characteristics': {
                'data_volume_gb': {'low': 100, 'medium': 1000, 'high': 10000},
                'transaction_volume': {'low': 1000, 'medium': 10000, 'high': 100000}
            }
        }
    
    def assess_migration_complexity(self, oracle_metadata):
        """Assess migration complexity based on Oracle database metadata"""
        complexity_score = 0
        assessment_details = {}
        
        # Schema complexity assessment
        table_count = oracle_metadata.get('table_count', 0)
        index_count = oracle_metadata.get('index_count', 0)
        constraint_count = oracle_metadata.get('constraint_count', 0)
        
        schema_complexity = self.calculate_complexity_level(
            table_count, self.complexity_factors['schema_complexity']['table_count']
        )
        
        assessment_details['schema_complexity'] = {
            'level': schema_complexity,
            'table_count': table_count,
            'index_count': index_count,
            'constraint_count': constraint_count
        }
        
        # Feature usage assessment
        oracle_features_used = oracle_metadata.get('oracle_features', [])
        feature_complexity = len(oracle_features_used)
        assessment_details['feature_complexity'] = {
            'features_count': feature_complexity,
            'features_used': oracle_features_used,
            'conversion_difficulty': self.assess_feature_conversion_difficulty(oracle_features_used)
        }
        
        # Data characteristics
        data_volume = oracle_metadata.get('data_volume_gb', 0)
        transaction_volume = oracle_metadata.get('daily_transactions', 0)
        data_complexity = self.calculate_complexity_level(
            data_volume, self.complexity_factors['data_characteristics']['data_volume_gb']
        )
        
        assessment_details['data_complexity'] = {
            'level': data_complexity,
            'data_volume_gb': data_volume,
            'daily_transactions': transaction_volume
        }
        
        # Calculate overall complexity score
        complexity_score = (
            self.get_complexity_score(schema_complexity) * 0.4 +
            feature_complexity * 2 + # Each feature adds complexity
            self.get_complexity_score(data_complexity) * 0.3
        )
        
        return {
            'overall_complexity_score': complexity_score,
            'complexity_level': self.determine_overall_complexity(complexity_score),
            'assessment_details': assessment_details,
            'recommendations': self.generate_migration_recommendations(complexity_score, assessment_details)
        }
    
    def calculate_complexity_level(self, value, thresholds):
        """Calculate complexity level based on value and thresholds"""
        if value <= thresholds['low']:
            return 'low'
        elif value <= thresholds['medium']:
            return 'medium'
        else:
            return 'high'
    
    def assess_feature_conversion_difficulty(self, features):
        """Assess difficulty of converting Oracle-specific features"""
        difficulty_mapping = {
            'partitioning': 'medium',
            'materialized_views': 'low',
            'packages': 'high',
            'triggers': 'medium',
            'functions': 'medium',
            'sequences': 'low',
            'synonyms': 'low',
            'xml_data_types': 'high',
            'spatial_data': 'medium',
            'advanced_security': 'high'
        }
        
        conversion_analysis = {}
        for feature in features:
            conversion_analysis[feature] = difficulty_mapping.get(feature, 'medium')
        
        return conversion_analysis

# Oracle to PostgreSQL Migration Project Plan
migration_project:
  phases:
    phase_1_assessment:
      duration_weeks: 4
      deliverables:
        - database_inventory
        - complexity_analysis
        - application_dependency_mapping
        - risk_assessment
        - migration_strategy_document
    
    phase_2_preparation:
      duration_weeks: 6
      deliverables:
        - schema_conversion_scripts
        - data_migration_pipeline
        - application_code_analysis
        - test_environment_setup
        - migration_tooling_setup
    
    phase_3_pilot_migration:
      duration_weeks: 4
      deliverables:
        - pilot_database_migration
        - performance_baseline
        - application_testing_results
        - process_refinement
    
    phase_4_production_migration:
      duration_weeks: 3
      deliverables:
        - production_cutover
        - performance_validation
        - application_verification
        - go_live_support
    
    phase_5_optimization:
      duration_weeks: 4
      deliverables:
        - performance_tuning
        - monitoring_setup
        - documentation_completion
        - knowledge_transfer
  
  resource_requirements:
    database_architect: 1.0
    database_administrator: 2.0
    application_developer: 2.0
    test_engineer: 1.0
    project_manager: 0.5
  
  success_criteria:
    functional:
      - zero_data_loss: true
      - application_functionality_preserved: 100%
      - data_integrity_maintained: true
    performance:
      - response_time_degradation: < 10%
      - throughput_maintenance: >= 95%
      - availability_target: 99.9%
    business:
      - migration_window_met: true
      - budget_adherence: 100%
      - timeline_adherence: 95%

class OracleToPostgreSQLMapper:
    def __init__(self):
        self.data_type_mapping = {
            # Numeric types
            'NUMBER': self.map_number_type,
            'NUMBER(p,s)': self.map_number_with_precision,
            'INTEGER': 'INTEGER',
            'INT': 'INTEGER',
            'SMALLINT': 'SMALLINT',
            'FLOAT': 'DOUBLE PRECISION',
            'BINARY_FLOAT': 'REAL',
            'BINARY_DOUBLE': 'DOUBLE PRECISION',
            
            # Character types
            'VARCHAR2': self.map_varchar2,
            'CHAR': self.map_char,
            'NCHAR': self.map_nchar,
            'NVARCHAR2': self.map_nvarchar2,
            'CLOB': 'TEXT',
            'NCLOB': 'TEXT',
            'LONG': 'TEXT',
            
            # Date and time types
            'DATE': 'TIMESTAMP(0)',
            'TIMESTAMP': self.map_timestamp,
            'TIMESTAMP WITH TIME ZONE': 'TIMESTAMP WITH TIME ZONE',
            'TIMESTAMP WITH LOCAL TIME ZONE': 'TIMESTAMP WITH TIME ZONE',
            'INTERVAL YEAR TO MONTH': 'INTERVAL',
            'INTERVAL DAY TO SECOND': 'INTERVAL',
            
            # Binary types
            'RAW': 'BYTEA',
            'BLOB': 'BYTEA',
            'BFILE': self.map_bfile,
            
            # Rowid
            'ROWID': 'TEXT',
            'UROWID': 'TEXT'
        }
        
        self.constraint_mapping = {
            'PRIMARY KEY': 'PRIMARY KEY',
            'FOREIGN KEY': 'FOREIGN KEY',
            'UNIQUE': 'UNIQUE',
            'CHECK': 'CHECK',
            'NOT NULL': 'NOT NULL'
        }
    
    def convert_schema(self, oracle_schema):
        """Convert Oracle schema to PostgreSQL equivalent"""
        postgresql_schema = {
            'tables': [],
            'indexes': [],
            'constraints': [],
            'sequences': [],
            'functions': [],
            'triggers': [],
            'conversion_notes': []
        }
        
        # Convert tables
        for table in oracle_schema.get('tables', []):
            converted_table = self.convert_table(table)
            postgresql_schema['tables'].append(converted_table)
        
        # Convert indexes
        for index in oracle_schema.get('indexes', []):
            converted_index = self.convert_index(index)
            postgresql_schema['indexes'].append(converted_index)
        
        # Convert sequences
        for sequence in oracle_schema.get('sequences', []):
            converted_sequence = self.convert_sequence(sequence)
            postgresql_schema['sequences'].append(converted_sequence)
        
        return postgresql_schema
    
    def convert_table(self, oracle_table):
        """Convert Oracle table definition to PostgreSQL"""
        table_name = oracle_table['table_name']
        columns = []
        
        for column in oracle_table['columns']:
            converted_column = self.convert_column(column)
            columns.append(converted_column)
        
        # Handle table-level constraints
        constraints = []
        for constraint in oracle_table.get('constraints', []):
            converted_constraint = self.convert_constraint(constraint)
            constraints.append(converted_constraint)
        
        return {
            'table_name': table_name,
            'columns': columns,
            'constraints': constraints,
            'conversion_notes': []
        }
    
    def convert_column(self, oracle_column):
        """Convert Oracle column definition to PostgreSQL"""
        column_name = oracle_column['column_name']
        oracle_type = oracle_column['data_type']
        nullable = oracle_column.get('nullable', 'Y') == 'Y'
        default_value = oracle_column.get('default_value')
        
        # Map data type
        if oracle_type in self.data_type_mapping:
            mapping_func = self.data_type_mapping[oracle_type]
            if callable(mapping_func):
                postgresql_type = mapping_func(oracle_column)
            else:
                postgresql_type = mapping_func
        else:
            postgresql_type = 'TEXT' # Default fallback
        
        return {
            'column_name': column_name,
            'data_type': postgresql_type,
            'nullable': nullable,
            'default_value': self.convert_default_value(default_value),
            'original_oracle_type': oracle_type
        }
    
    def map_number_type(self, column_info):
        """Map Oracle NUMBER type to appropriate PostgreSQL type"""
        precision = column_info.get('precision')
        scale = column_info.get('scale')
        
        if precision is None and scale is None:
            return 'NUMERIC'
        elif scale == 0:
            if precision <= 4:
                return 'SMALLINT'
            elif precision <= 9:
                return 'INTEGER'
            elif precision <= 18:
                return 'BIGINT'
            else:
                return f'NUMERIC({precision})'
        else:
            return f'NUMERIC({precision},{scale})'
    
    def map_varchar2(self, column_info):
        """Map Oracle VARCHAR2 to PostgreSQL VARCHAR"""
        length = column_info.get('length', 4000)
        return f'VARCHAR({length})'
    
    def map_timestamp(self, column_info):
        """Map Oracle TIMESTAMP to PostgreSQL TIMESTAMP"""
        precision = column_info.get('precision', 6)
        return f'TIMESTAMP({precision})'

-- Oracle to PostgreSQL index conversion examples

-- Oracle bitmap index (not directly supported in PostgreSQL)
-- Oracle: CREATE BITMAP INDEX idx_status ON orders (status);
-- PostgreSQL: Use partial indexes or GIN indexes
CREATE INDEX idx_status_active ON orders (customer_id) WHERE status = 'ACTIVE';
CREATE INDEX idx_status_gin ON orders USING GIN (status) WHERE status IN ('ACTIVE', 'PENDING');

-- Oracle function-based index
-- Oracle: CREATE INDEX idx_upper_name ON customers (UPPER(name));
-- PostgreSQL: Similar syntax supported
CREATE INDEX idx_upper_name ON customers (UPPER(name));

-- Oracle reverse key index
-- Oracle: CREATE INDEX idx_id_reverse ON orders (id) REVERSE;
-- PostgreSQL: Not directly supported, consider hash index
CREATE INDEX idx_id_hash ON orders USING HASH (id);

-- Complex constraint conversion
-- Oracle: Deferrable constraints
ALTER TABLE order_items 
ADD CONSTRAINT fk_order_id 
FOREIGN KEY (order_id) REFERENCES orders(id) 
DEFERRABLE INITIALLY DEFERRED;

-- PostgreSQL: Same syntax supported
ALTER TABLE order_items 
ADD CONSTRAINT fk_order_id 
FOREIGN KEY (order_id) REFERENCES orders(id) 
DEFERRABLE INITIALLY DEFERRED;

class MigrationStrategySelector:
    def __init__(self):
        self.strategies = {
            'big_bang': {
                'description': 'Complete migration during single maintenance window',
                'downtime': 'high',
                'complexity': 'low',
                'rollback_difficulty': 'high',
                'suitable_for': ['small_databases', 'flexible_downtime']
            },
            'phased_migration': {
                'description': 'Migrate tables/modules in phases',
                'downtime': 'medium',
                'complexity': 'medium',
                'rollback_difficulty': 'medium',
                'suitable_for': ['medium_databases', 'modular_applications']
            },
            'parallel_run': {
                'description': 'Run Oracle and PostgreSQL in parallel',
                'downtime': 'low',
                'complexity': 'high',
                'rollback_difficulty': 'low',
                'suitable_for': ['large_databases', 'mission_critical']
            },
            'logical_replication': {
                'description': 'Use logical replication for continuous sync',
                'downtime': 'minimal',
                'complexity': 'high',
                'rollback_difficulty': 'low',
                'suitable_for': ['zero_downtime_required', 'large_databases']
            }
        }
    
    def recommend_strategy(self, database_size_gb, max_downtime_hours, 
                          complexity_level, criticality):
        """Recommend optimal migration strategy"""
        recommendations = []
        
        for strategy_name, strategy_info in self.strategies.items():
            score = self.calculate_strategy_score(
                strategy_name, database_size_gb, max_downtime_hours, 
                complexity_level, criticality
            )
            recommendations.append({
                'strategy': strategy_name,
                'score': score,
                'description': strategy_info['description'],
                'pros_cons': self.get_strategy_pros_cons(strategy_name)
            })
        
        # Sort by score
        recommendations.sort(key=lambda x: x['score'], reverse=True)
        
        return {
            'recommended_strategy': recommendations[0],
            'all_options': recommendations,
            'decision_factors': {
                'database_size_gb': database_size_gb,
                'max_downtime_hours': max_downtime_hours,
                'complexity_level': complexity_level,
                'criticality': criticality
            }
        }
    
    def calculate_strategy_score(self, strategy, db_size, downtime, complexity, criticality):
        """Calculate strategy suitability score"""
        score = 50  # Base score
        
        # Adjust based on database size
        if strategy == 'big_bang' and db_size > 1000:
            score -= 30  # Big bang not suitable for large databases
        elif strategy == 'logical_replication' and db_size < 100:
            score -= 10  # May be overkill for small databases
        
        # Adjust based on downtime requirements
        if downtime < 2:  # Less than 2 hours
            if strategy in ['parallel_run', 'logical_replication']:
                score += 20
            elif strategy == 'big_bang':
                score -= 40
        
        # Adjust based on complexity
        if complexity == 'high':
            if strategy in ['big_bang', 'phased_migration']:
                score += 10  # Simpler strategies better for complex schemas
            else:
                score -= 15
        
        # Adjust based on criticality
        if criticality == 'mission_critical':
            if strategy in ['parallel_run', 'logical_replication']:
                score += 15
            elif strategy == 'big_bang':
                score -= 25
        
        return max(0, min(100, score))  # Ensure score is between 0-100

class DataMigrationToolkit:
    def __init__(self):
        self.tools = {
            'ora2pg': {
                'type': 'schema_and_data',
                'strengths': ['comprehensive', 'mature', 'free'],
                'weaknesses': ['complex_setup', 'limited_parallel_processing'],
                'best_for': 'complete_migrations'
            },
            'aws_dms': {
                'type': 'data_migration_service',
                'strengths': ['managed_service', 'minimal_downtime', 'monitoring'],
                'weaknesses': ['cloud_dependent', 'cost', 'limited_transformation'],
                'best_for': 'cloud_migrations'
            },
            'pentaho_kettle': {
                'type': 'etl_tool',
                'strengths': ['visual_interface', 'transformation_capabilities', 'scheduling'],
                'weaknesses': ['resource_intensive', 'learning_curve'],
                'best_for': 'complex_transformations'
            },
            'custom_scripts': {
                'type': 'scripted_migration',
                'strengths': ['full_control', 'optimization', 'integration'],
                'weaknesses': ['development_time', 'maintenance'],
                'best_for': 'specific_requirements'
            }
        }
    
    def generate_ora2pg_config(self, oracle_connection, postgresql_connection, migration_options):
        """Generate ora2pg configuration file"""
        config = f"""
# Ora2Pg configuration file for Oracle to PostgreSQL migration

# Oracle database connection
ORACLE_DSN dbi:Oracle:host={oracle_connection['host']};sid={oracle_connection['sid']};port={oracle_connection['port']}
ORACLE_USER {oracle_connection['username']}
ORACLE_PWD {oracle_connection['password']}

# PostgreSQL database connection 
PG_DSN dbi:Pg:dbname={postgresql_connection['database']};host={postgresql_connection['host']};port={postgresql_connection['port']}
PG_USER {postgresql_connection['username']}
PG_PWD {postgresql_connection['password']}

# Migration options
TYPE {migration_options.get('type', 'TABLE')}
OUTPUT {migration_options.get('output_file', 'output.sql')}
OUTPUT_DIR {migration_options.get('output_dir', './migration_output')}

# Schema options
SCHEMA {migration_options.get('schema', 'PUBLIC')}
CREATE_SCHEMA {migration_options.get('create_schema', '1')}
COMPILE_SCHEMA {migration_options.get('compile_schema', '1')}

# Data export options
DATA_LIMIT {migration_options.get('data_limit', '0')}
PARALLEL_TABLES {migration_options.get('parallel_tables', '4')}
JOBS {migration_options.get('jobs', '4')}

# Object filters
ALLOW {migration_options.get('allow_objects', 'TABLE,VIEW,SEQUENCE,FUNCTION,PROCEDURE')}
EXCLUDE {migration_options.get('exclude_objects', '')}

# Performance options
CHUNK_SIZE {migration_options.get('chunk_size', '10000')}
ORACLE_COPIES {migration_options.get('oracle_copies', '2')}

# Conversion options
PG_NUMERIC_TYPE {migration_options.get('pg_numeric_type', '1')}
DEFAULT_NUMERIC {migration_options.get('default_numeric', 'BIGINT')}
PG_INTEGER_TYPE {migration_options.get('pg_integer_type', '1')}

# Debug and logging
DEBUG {migration_options.get('debug', '0')}
QUIET {migration_options.get('quiet', '0')}
LOGFILE {migration_options.get('logfile', 'ora2pg.log')}
"""
        return config

-- Data validation queries for Oracle to PostgreSQL migration

-- Row count validation
-- Oracle
SELECT table_name, num_rows 
FROM user_tables 
WHERE table_name IN ('CUSTOMERS', 'ORDERS', 'ORDER_ITEMS')
ORDER BY table_name;

-- PostgreSQL
SELECT 
    schemaname,
    tablename,
    n_tup_ins - n_tup_del as estimated_rows
FROM pg_stat_user_tables 
WHERE tablename IN ('customers', 'orders', 'order_items')
ORDER BY tablename;

-- Data integrity validation
-- Check for NULL values in key columns
SELECT 
    'customers' as table_name,
    COUNT(*) as total_rows,
    COUNT(customer_id) as non_null_customer_id,
    COUNT(email) as non_null_email
FROM customers
UNION ALL
SELECT 
    'orders' as table_name,
    COUNT(*) as total_rows,
    COUNT(order_id) as non_null_order_id,
    COUNT(customer_id) as non_null_customer_id
FROM orders;

-- Data type validation
-- Numeric precision validation
SELECT 
    column_name,
    data_type,
    numeric_precision,
    numeric_scale,
    COUNT(*) as column_count
FROM information_schema.columns 
WHERE table_name = 'financial_data'
    AND data_type IN ('numeric', 'decimal')
GROUP BY column_name, data_type, numeric_precision, numeric_scale;

-- Date range validation
SELECT 
    MIN(order_date) as min_date,
    MAX(order_date) as max_date,
    COUNT(*) as total_records,
    COUNT(DISTINCT DATE_TRUNC('month', order_date)) as months_span
FROM orders;

-- Foreign key validation
SELECT 
    o.order_id,
    o.customer_id,
    c.customer_id as customer_exists
FROM orders o
LEFT JOIN customers c ON o.customer_id = c.customer_id
WHERE c.customer_id IS NULL;

-- Character encoding validation
SELECT 
    customer_id,
    name,
    LENGTH(name) as name_length,
    OCTET_LENGTH(name) as name_bytes
FROM customers 
WHERE OCTET_LENGTH(name) != LENGTH(name) -- Multi-byte characters
LIMIT 10;