NoSQL vs SQL: A Comprehensive Comparison

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

# Migration Assessment Checklist
 
## Current State Analysis
- [ ] Document all tables/collections, row counts, data sizes
- [ ] Map all data types and their target equivalents
- [ ] Identify relationships and integrity constraints
- [ ] Catalog stored procedures, triggers, views
- [ ] List all application queries and access patterns
- [ ] Document indexes and their usage frequency
- [ ] Identify sensitive data requiring special handling
- [ ] Measure current performance baseline (latency, throughput)
 
## Target State Definition
- [ ] Define target database schema/data model
- [ ] Plan data type mappings (e.g., SERIAL → UUID)
- [ ] Decide on denormalization strategy if applicable
- [ ] Design new indexes for target query patterns
- [ ] Plan for features not available in target (stored procs → app code)
 
## Dependency Mapping
- [ ] List all applications connecting to current database
- [ ] Identify third-party tools (BI, CDC, backups) requiring changes
- [ ] Map team responsibilities and skills
- [ ] Identify compliance/audit requirements
- [ ] Document SLAs that must be maintained during migration
 
## Risk Assessment
- [ ] Estimate data loss window acceptable to business
- [ ] Identify rollback scenarios and procedures
- [ ] Plan for extended rollback period
- [ ] Define go/no-go criteria for final cutover

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71

// SQL Schema (normalized)
/*
CREATE TABLE customers (id, name, email);
CREATE TABLE addresses (id, customer_id, street, city, country, is_primary);
CREATE TABLE orders (id, customer_id, order_date, status);
CREATE TABLE order_items (order_id, product_id, quantity, price);
CREATE TABLE products (id, name, price, category_id);
CREATE TABLE categories (id, name, parent_id);
*/
 
// Document Schema (denormalized for read optimization)
// Decision: What to embed vs reference?
 
// Customer document - embed addresses (1:few, queried together)
{
    "_id": ObjectId("..."),
    "name": "Alice Smith",
    "email": "alice@example.com",
    "addresses": [
        {
            "street": "123 Main St",
            "city": "Seattle",
            "country": "USA",
            "isPrimary": true
        },
        {
            "street": "456 Work Ave",
            "city": "Seattle", 
            "country": "USA",
            "isPrimary": false
        }
    ]
}
 
// Order document - embed items, reference customer and products
// Items embedded (always queried with order)
// Customer referenced (avoid duplication, infrequent access)
// Products referenced with denormalized fields (name, price at time of order)
{
    "_id": ObjectId("..."),
    "customerId": ObjectId("..."),      // Reference
    "customerEmail": "alice@example.com", // Denormalized for notifications
    "orderDate": ISODate("2024-01-15"),
    "status": "shipped",
    "items": [                           // Embedded
        {
            "productId": ObjectId("..."),  // Reference
            "productName": "Widget X",     // Denormalized (snapshot at order time)
            "quantity": 2,
            "unitPrice": 29.99,            // Price at time of order
            "lineTotal": 59.98
        }
    ],
    "shippingAddress": {                 // Embedded (snapshot at order time)
        "street": "123 Main St",
        "city": "Seattle",
        "country": "USA"
    },
    "totals": {
        "subtotal": 59.98,
        "tax": 6.00,
        "shipping": 5.99,
        "total": 71.97
    }
}
 
// Transformation considerations:
// 1. What's queried together? → Embed
// 2. What changes independently? → Reference
// 3. What needs historical snapshot? → Embed with copy
// 4. What would create unbounded arrays? → Reference

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82

-- MongoDB document (denormalized)
/*
{
    "_id": ObjectId("..."),
    "title": "Understanding Databases",
    "author": {
        "name": "Jane Expert",
        "bio": "Database specialist...",
        "social": {
            "twitter": "@janedb",
            "github": "janeexpert"
        }
    },
    "tags": ["databases", "nosql", "sql"],
    "comments": [
        {
            "user": "reader1",
            "text": "Great article!",
            "timestamp": ISODate("2024-01-15T10:30:00Z"),
            "likes": 15
        },
        {
            "user": "reader2",
            "text": "Very helpful",
            "timestamp": ISODate("2024-01-15T11:45:00Z"),
            "likes": 8
        }
    ]
}
*/
 
-- Normalized SQL Schema
CREATE TABLE authors (
    id SERIAL PRIMARY KEY,
    name VARCHAR(255) NOT NULL,
    bio TEXT,
    twitter_handle VARCHAR(100),
    github_handle VARCHAR(100)
);
 
CREATE TABLE articles (
    id SERIAL PRIMARY KEY,
    mongo_id VARCHAR(24) UNIQUE,  -- Preserve original ID for mapping
    title VARCHAR(500) NOT NULL,
    author_id INTEGER REFERENCES authors(id),
    created_at TIMESTAMP DEFAULT NOW()
);
 
CREATE TABLE tags (
    id SERIAL PRIMARY KEY,
    name VARCHAR(100) UNIQUE NOT NULL
);
 
CREATE TABLE article_tags (
    article_id INTEGER REFERENCES articles(id) ON DELETE CASCADE,
    tag_id INTEGER REFERENCES tags(id) ON DELETE CASCADE,
    PRIMARY KEY (article_id, tag_id)
);
 
CREATE TABLE comments (
    id SERIAL PRIMARY KEY,
    article_id INTEGER REFERENCES articles(id) ON DELETE CASCADE,
    username VARCHAR(100) NOT NULL,
    content TEXT NOT NULL,
    created_at TIMESTAMP NOT NULL,
    likes INTEGER DEFAULT 0
);
 
-- Migration queries
-- 1. Extract unique authors
INSERT INTO authors (name, bio, twitter_handle, github_handle)
SELECT DISTINCT 
    doc->>'author.name',
    doc->>'author.bio',
    doc->'author'->'social'->>'twitter',
    doc->'author'->'social'->>'github'
FROM mongo_import;
 
-- 2. Insert articles with author lookup
-- 3. Extract and insert unique tags
-- 4. Create article-tag relationships
-- 5. Unnest comments array into comments table

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85

// Dual Write Pattern with Shadow Read Comparison
 
class MigratingOrderRepository {
    private oldDb: PostgresClient;
    private newDb: MongoClient;
    private metrics: MetricsClient;
    private featureFlags: FeatureFlags;
    
    async save(order: Order): Promise<void> {
        // Always write to old database (source of truth during migration)
        await this.oldDb.query(
            'INSERT INTO orders (...) VALUES (...)',
            [order.id, order.customerId, ...]
        );
        
        // Also write to new database (async, non-blocking)
        this.writeToNewDb(order).catch(err => {
            this.metrics.increment('migration.dual_write.new_db_error');
            console.error('New DB write failed:', err);
            // Don't fail the operation - old DB is source of truth
        });
    }
    
    async findById(orderId: string): Promise<Order | null> {
        // Read from old database (source of truth)
        const oldResult = await this.oldDb.query(
            'SELECT * FROM orders WHERE id = $1',
            [orderId]
        );
        
        // Shadow read from new database (async, for comparison)
        if (this.featureFlags.isEnabled('migration.shadow_reads')) {
            this.shadowRead(orderId, oldResult.rows[0]).catch(err => {
                this.metrics.increment('migration.shadow_read.error');
            });
        }
        
        return oldResult.rows[0] || null;
    }
    
    private async shadowRead(orderId: string, oldResult: any): Promise<void> {
        const newResult = await this.newDb.collection('orders')
            .findOne({ _id: orderId });
        
        // Compare results
        const differences = this.compare(oldResult, newResult);
        
        if (differences.length > 0) {
            this.metrics.increment('migration.shadow_read.mismatch');
            console.warn('Data mismatch:', { orderId, differences });
            // Log for investigation, don't fail
        } else {
            this.metrics.increment('migration.shadow_read.match');
        }
    }
    
    private compare(old: any, new_: any): string[] {
        // Implement comparison logic accounting for schema differences
        // Return list of field differences
    }
}
 
// Gradually shift traffic
class MigratingOrderRepositoryV2 {
    async findById(orderId: string): Promise<Order | null> {
        const readFromNew = this.featureFlags.percentage('migration.read_new_db');
        
        if (Math.random() < readFromNew) {
            // Read from new database for this percentage of requests
            const result = await this.newDb.collection('orders')
                .findOne({ _id: orderId });
                
            if (result) {
                this.metrics.increment('migration.read.new_db.success');
                return this.mapToOrder(result);
            } else {
                // Fallback to old if not found (consistency lag)
                this.metrics.increment('migration.read.new_db.fallback');
                return this.readFromOldDb(orderId);
            }
        }
        
        return this.readFromOldDb(orderId);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50

# CDC-based migration pipeline
 
# Phase 1: Initial bulk load
initial_load:
  source: PostgreSQL
  target: MongoDB
  method: pg_dump | transform | mongoimport
  estimated_time: "4 hours for 500GB"
  
# Phase 2: Continuous sync via CDC
cdc_pipeline:
  source:
    type: PostgreSQL
    connector: Debezium
    tables:
      - public.orders
      - public.order_items
      - public.customers
      
  transform:
    type: Kafka Streams
    logic: |
      - Join order with items
      - Embed customer address
      - Convert types (SERIAL → ObjectId)
      - Transform timestamps
      
  target:
    type: MongoDB
    connector: MongoDB Kafka Connector
    collection: orders
    write_mode: upsert
    
# Phase 3: Cutover
cutover:
  steps:
    1: Stop writes to old database
    2: Wait for CDC lag to reach zero
    3: Verify row counts match
    4: Switch application to read/write new database
    5: Monitor for 24 hours
    6: Decommission old database (after 30-day safety window)
    
rollback:
  if_within: "30 days"
  method: |
    - Stop writes to new database
    - Bulk export and import to old
    - Resume CDC in reverse direction
    - Switch application back

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62

// Abstract repository interface - database agnostic
interface OrderRepository {
    findById(id: string): Promise<Order | null>;
    findByCustomer(customerId: string, limit: number): Promise<Order[]>;
    save(order: Order): Promise<void>;
    findPendingOrders(olderThanDays: number): Promise<Order[]>;
}
 
// PostgreSQL implementation
class PostgresOrderRepository implements OrderRepository {
    async findById(id: string): Promise<Order | null> {
        const result = await this.pool.query(
            `SELECT o.*, c.name as customer_name, c.email as customer_email
             FROM orders o
             JOIN customers c ON o.customer_id = c.id
             WHERE o.id = $1`,
            [id]
        );
        return result.rows[0] ? this.mapToOrder(result.rows[0]) : null;
    }
    
    // ... other methods
}
 
// MongoDB implementation
class MongoOrderRepository implements OrderRepository {
    async findById(id: string): Promise<Order | null> {
        const doc = await this.collection.findOne({ _id: new ObjectId(id) });
        return doc ? this.mapToOrder(doc) : null;
    }
    
    // ... other methods
}
 
// Migration-aware factory
function createOrderRepository(config: Config): OrderRepository {
    switch (config.database.migration_phase) {
        case 'postgres_only':
            return new PostgresOrderRepository(config.postgres);
            
        case 'dual_write':
            return new DualWriteOrderRepository(
                new PostgresOrderRepository(config.postgres),
                new MongoOrderRepository(config.mongo)
            );
            
        case 'mongo_primary':
            return new MongoOrderRepository(config.mongo);
            
        default:
            throw new Error('Unknown migration phase');
    }
}
 
// Application code remains unchanged throughout migration
class OrderService {
    constructor(private orderRepo: OrderRepository) {}
    
    async getOrder(id: string): Promise<Order | null> {
        return this.orderRepo.findById(id);
    }
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97

# Data validation framework for migration
 
import hashlib
from typing import Dict, List
 
class MigrationValidator:
    def __init__(self, source_db, target_db):
        self.source = source_db
        self.target = target_db
        self.errors: List[str] = []
    
    def validate_row_counts(self) -> bool:
        """Verify all entities migrated"""
        tables = [
            ('customers', 'customers'),
            ('orders', 'orders'),
            ('products', 'products'),
        ]
        
        for source_table, target_collection in tables:
            source_count = self.source.execute(
                f"SELECT COUNT(*) FROM {source_table}"
            ).fetchone()[0]
            
            target_count = self.target[target_collection].count_documents({})
            
            if source_count != target_count:
                self.errors.append(
                    f"{source_table}: source={source_count}, target={target_count}"
                )
        
        return len(self.errors) == 0
    
    def validate_data_integrity(self, sample_size: int = 10000) -> bool:
        """Spot check data values match"""
        # Random sample of IDs from source
        sample_ids = self.source.execute(
            f"SELECT id FROM orders ORDER BY RANDOM() LIMIT {sample_size}"
        ).fetchall()
        
        mismatches = []
        for (order_id,) in sample_ids:
            source_order = self.fetch_source_order(order_id)
            target_order = self.fetch_target_order(order_id)
            
            diffs = self.compare_orders(source_order, target_order)
            if diffs:
                mismatches.append({'id': order_id, 'diffs': diffs})
        
        if mismatches:
            self.errors.extend([
                f"Data mismatch: {m['id']}: {m['diffs']}" 
                for m in mismatches[:10]  # Log first 10
            ])
        
        mismatch_rate = len(mismatches) / sample_size
        print(f"Mismatch rate: {mismatch_rate:.4%}")
        return mismatch_rate < 0.0001  # < 0.01% acceptable
    
    def validate_checksums(self) -> bool:
        """Verify critical numeric fields sum correctly"""
        checks = [
            # (source_query, target_pipeline, field_name)
            (
                "SELECT SUM(total) FROM orders WHERE status = 'completed'",
                [{"$match": {"status": "completed"}}, 
                 {"$group": {"_id": None, "sum": {"$sum": "$total"}}}],
                "completed_order_total"
            ),
        ]
        
        for source_query, target_pipeline, field_name in checks:
            source_sum = self.source.execute(source_query).fetchone()[0]
            target_result = list(self.target.orders.aggregate(target_pipeline))
            target_sum = target_result[0]['sum'] if target_result else 0
            
            # Allow for floating point precision
            if abs(source_sum - target_sum) > 0.01:
                self.errors.append(
                    f"{field_name}: source={source_sum}, target={target_sum}"
                )
        
        return len(self.errors) == 0
 
# Run validation suite
validator = MigrationValidator(postgres_conn, mongo_client)
results = {
    'row_counts': validator.validate_row_counts(),
    'data_integrity': validator.validate_data_integrity(),
    'checksums': validator.validate_checksums(),
}
 
if all(results.values()):
    print("✅ Migration validation passed")
else:
    print("❌ Migration validation failed")
    print("Errors:", validator.errors)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58

# Migration Cutover Runbook
 
## Pre-Cutover (T-24 hours)
- [ ] All stakeholders notified
- [ ] Support team briefed on potential issues
- [ ] Rollback procedure tested in staging
- [ ] All validation tests passing
- [ ] CDC lag < 10 seconds consistently
- [ ] Go/no-go meeting completed
 
## Pre-Cutover (T-1 hour)
- [ ] Freeze all deployments
- [ ] Confirm team availability for duration
- [ ] Open communication channel (Slack, bridge call)
- [ ] Prepare status update templates
 
## Cutover Procedure
 
### Step 1: Stop Writes (T+0:00)
- [ ] Enable maintenance mode / stop application writes
- [ ] Verify no new writes in old database (check WAL position)
- Record time: _____________
 
### Step 2: Final Sync (T+0:05)
- [ ] Wait for CDC lag to reach zero
- [ ] Run final validation (row counts, checksums)
- [ ] If validation fails → ABORT and rollback
- Record time: _____________
 
### Step 3: Switch Configuration (T+0:15)
- [ ] Deploy application config pointing to new database
- [ ] Verify application connecting to new database
- [ ] Enable writes to new database
- Record time: _____________
 
### Step 4: Verification (T+0:20)
- [ ] Smoke test critical paths manually
- [ ] Verify metrics (error rates, latencies)
- [ ] Check all services healthy
- Record time: _____________
 
### Step 5: Re-enable Traffic (T+0:30)
- [ ] Disable maintenance mode
- [ ] Monitor closely for 30 minutes
- [ ] If issues → ROLLBACK
- Record time: _____________
 
## Success Criteria (before declaring victory)
- [ ] Error rate < 0.1% for 30 minutes
- [ ] p99 latency within 20% of baseline
- [ ] All critical transactions completing
- [ ] No data integrity alerts
 
## Rollback Trigger Conditions
- Error rate > 1%
- Critical transaction failures
- Data corruption detected
- Latency > 3x baseline for > 5 minutes