Opinion: Why You Should Ditch MongoDB 8 and Use PostgreSQL 18 for All Your Data Storage Needs

import psycopg
import pymongo
import time
import logging
import os
from dataclasses import dataclass
from typing import List, Dict
import random
import string

# Configure logging
logging.basicConfig(
    level=logging.INFO,
    format=\"%(asctime)s - %(levelname)s - %(message)s\"
)
logger = logging.getLogger(__name__)

@dataclass
class BenchmarkConfig:
    pg_conn_str: str
    mongo_uri: str
    database: str
    collection: str
    num_operations: int = 100_000
    document_size_kb: int = 1
    concurrency: int = 16

def generate_document(size_kb: int) -> Dict:
    \"\"\"Generate a random document of approximately size_kb kilobytes\"\"\"
    base_doc = {
        \"user_id\": random.randint(1, 1_000_000),
        \"timestamp\": time.time(),
        \"event_type\": random.choice([\"click\", \"view\", \"purchase\", \"logout\"]),
        \"metadata\": \"\".join(random.choices(string.ascii_letters + string.digits, k=size_kb * 1024 - 100))
    }
    return base_doc

def benchmark_postgresql(config: BenchmarkConfig) -> Dict:
    \"\"\"Run YCSB Workload A (50% insert, 50% update) against PostgreSQL 18\"\"\"
    try:
        with psycopg.connect(config.pg_conn_str) as conn:
            with conn.cursor() as cur:
                # Create table with JSONB column if not exists
                cur.execute(\"\"\"
                    CREATE TABLE IF NOT EXISTS ycsb_workload_a (
                        id SERIAL PRIMARY KEY,
                        doc JSONB NOT NULL,
                        created_at TIMESTAMPTZ DEFAULT NOW()
                    );
                    CREATE INDEX IF NOT EXISTS idx_ycsb_doc_gin ON ycsb_workload_a USING GIN (doc);
                \"\"\")
                conn.commit()

                # Pre-generate documents
                docs = [generate_document(config.document_size_kb) for _ in range(config.num_operations // 2)]
                insert_times = []
                update_times = []

                # Run insert workload
                for i, doc in enumerate(docs):
                    start = time.perf_counter()
                    cur.execute(
                        \"INSERT INTO ycsb_workload_a (doc) VALUES (%s) RETURNING id;\",
                        (psycopg.types.json.Json(doc),)
                    )
                    conn.commit()
                    insert_times.append(time.perf_counter() - start)

                # Run update workload (update random existing docs)
                cur.execute(\"SELECT id FROM ycsb_workload_a ORDER BY RANDOM() LIMIT %s;\", (config.num_operations // 2,))
                existing_ids = [row[0] for row in cur.fetchall()]
                for doc_id in existing_ids:
                    new_doc = generate_document(config.document_size_kb)
                    start = time.perf_counter()
                    cur.execute(
                        \"UPDATE ycsb_workload_a SET doc = %s WHERE id = %s;\",
                        (psycopg.types.json.Json(new_doc), doc_id)
                    )
                    conn.commit()
                    update_times.append(time.perf_counter() - start)

                total_time = sum(insert_times) + sum(update_times)
                throughput = config.num_operations / total_time
                p99_insert = sorted(insert_times)[int(len(insert_times) * 0.99)]
                p99_update = sorted(update_times)[int(len(update_times) * 0.99)]

                return {
                    \"db\": \"PostgreSQL 18\",
                    \"throughput_ops_sec\": round(throughput, 2),
                    \"p99_insert_ms\": round(p99_insert * 1000, 2),
                    \"p99_update_ms\": round(p99_update * 1000, 2)
                }
    except Exception as e:
        logger.error(f\"PostgreSQL benchmark failed: {e}\")
        raise

def benchmark_mongodb(config: BenchmarkConfig) -> Dict:
    \"\"\"Run YCSB Workload A (50% insert, 50% update) against MongoDB 8\"\"\"
    try:
        client = pymongo.MongoClient(config.mongo_uri)
        db = client[config.database]
        collection = db[config.collection]
        # Create index on _id (default, but ensure wiredTiger cache is configured)
        collection.create_index([(\"_id\", pymongo.ASCENDING)], background=True)

        docs = [generate_document(config.document_size_kb) for _ in range(config.num_operations // 2)]
        insert_times = []
        update_times = []

        # Insert workload
        for doc in docs:
            start = time.perf_counter()
            collection.insert_one(doc)
            insert_times.append(time.perf_counter() - start)

        # Update workload
        existing_docs = list(collection.aggregate([{\"$sample\": {\"size\": config.num_operations // 2}}]))
        for doc in existing_docs:
            new_doc = generate_document(config.document_size_kb)
            start = time.perf_counter()
            collection.update_one({\"_id\": doc[\"_id\"]}, {\"$set\": new_doc})
            update_times.append(time.perf_counter() - start)

        total_time = sum(insert_times) + sum(update_times)
        throughput = config.num_operations / total_time
        p99_insert = sorted(insert_times)[int(len(insert_times) * 0.99)]
        p99_update = sorted(update_times)[int(len(update_times) * 0.99)]

        return {
            \"db\": \"MongoDB 8\",
            \"throughput_ops_sec\": round(throughput, 2),
            \"p99_insert_ms\": round(p99_insert * 1000, 2),
            \"p99_update_ms\": round(p99_update * 1000, 2)
        }
    except Exception as e:
        logger.error(f\"MongoDB benchmark failed: {e}\")
        raise
    finally:
        client.close()

if __name__ == \"__main__\":
    config = BenchmarkConfig(
        pg_conn_str=os.getenv(\"PG_CONN_STR\", \"postgresql://postgres:postgres@localhost:5432/benchmark\"),
        mongo_uri=os.getenv(\"MONGO_URI\", \"mongodb://localhost:27017\"),
        database=\"benchmark\",
        collection=\"ycsb_a\"
    )

    logger.info(\"Starting PostgreSQL 18 benchmark...\")
    pg_results = benchmark_postgresql(config)
    logger.info(f\"PostgreSQL Results: {pg_results}\")

    logger.info(\"Starting MongoDB 8 benchmark...\")
    mongo_results = benchmark_mongodb(config)
    logger.info(f\"MongoDB Results: {mongo_results}\")

package main

import (
    \"context\"
    \"encoding/json\"
    \"fmt\"
    \"log\"
    \"os\"
    \"time\"

    \"github.com/jackc/pgx/v5\"
    \"github.com/jackc/pgx/v5/pgxpool\"
    \"go.mongodb.org/mongo-driver/v2/bson\"
    \"go.mongodb.org/mongo-driver/v2/mongo\"
    \"go.mongodb.org/mongo-driver/v2/mongo/options\"
    \"go.mongodb.org/mongo-driver/v2/mongo/readpref\"
)

const (
    mongoURI       = \"mongodb://localhost:27017\"
    postgresConnStr = \"postgresql://postgres:postgres@localhost:5432/migration\"
    batchSize      = 1000
)

// UserDoc represents a MongoDB document to migrate
type UserDoc struct {
    ID        bson.ObjectID `bson:\"_id\"`
    Username  string        `bson:\"username\"`
    Email     string        `bson:\"email\"`
    Metadata  bson.M        `bson:\"metadata\"`
    CreatedAt time.Time     `bson:\"created_at\"`
    UpdatedAt time.Time     `bson:\"updated_at\"`
}

// MigratedUser represents the PostgreSQL row structure
type MigratedUser struct {
    ID        string    `json:\"id\"`
    Username  string    `json:\"username\"`
    Email     string    `json:\"email\"`
    Metadata  []byte    `json:\"metadata\"` // JSONB stored as bytes
    CreatedAt time.Time `json:\"created_at\"`
    UpdatedAt time.Time `json:\"updated_at\"`
}

func main() {
    ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute)
    defer cancel()

    // Connect to MongoDB 8
    mongoClient, err := mongo.ConnectToServer(ctx, options.Client().ApplyURI(mongoURI))
    if err != nil {
        log.Fatalf(\"Failed to connect to MongoDB: %v\", err)
    }
    defer mongoClient.Disconnect(ctx)

    // Ping MongoDB to verify connection
    if err := mongoClient.Ping(ctx, readpref.Primary()); err != nil {
        log.Fatalf(\"Failed to ping MongoDB: %v\", err)
    }
    log.Println(\"Connected to MongoDB 8 successfully\")

    // Connect to PostgreSQL 18
    pgPool, err := pgxpool.New(ctx, postgresConnStr)
    if err != nil {
        log.Fatalf(\"Failed to connect to PostgreSQL: %v\", err)
    }
    defer pgPool.Close()
    log.Println(\"Connected to PostgreSQL 18 successfully\")

    // Create target table in PostgreSQL
    createTableSQL := `
        CREATE TABLE IF NOT EXISTS users (
            id TEXT PRIMARY KEY,
            username TEXT NOT NULL UNIQUE,
            email TEXT NOT NULL UNIQUE,
            metadata JSONB NOT NULL,
            created_at TIMESTAMPTZ NOT NULL,
            updated_at TIMESTAMPTZ NOT NULL
        );
        CREATE INDEX IF NOT EXISTS idx_users_metadata ON users USING GIN (metadata);
    `
    _, err = pgPool.Exec(ctx, createTableSQL)
    if err != nil {
        log.Fatalf(\"Failed to create PostgreSQL table: %v\", err)
    }

    // Fetch documents from MongoDB in batches
    collection := mongoClient.Database(\"app\").Collection(\"users\")
    cursor, err := collection.Find(ctx, bson.D{}, options.Find().SetBatchSize(batchSize))
    if err != nil {
        log.Fatalf(\"Failed to query MongoDB: %v\", err)
    }
    defer cursor.Close(ctx)

    var batch []MigratedUser
    processed := 0

    for cursor.Next(ctx) {
        var doc UserDoc
        if err := cursor.Decode(&doc); err != nil {
            log.Printf(\"Failed to decode MongoDB document: %v\", err)
            continue
        }

        // Convert metadata to JSON bytes for PostgreSQL JSONB
        metadataJSON, err := json.Marshal(doc.Metadata)
        if err != nil {
            log.Printf(\"Failed to marshal metadata: %v\", err)
            continue
        }

        batch = append(batch, MigratedUser{
            ID:        doc.ID.Hex(),
            Username:  doc.Username,
            Email:     doc.Email,
            Metadata:  metadataJSON,
            CreatedAt: doc.CreatedAt,
            UpdatedAt: doc.UpdatedAt,
        })

        // Insert batch when size is reached
        if len(batch) >= batchSize {
            if err := insertBatch(ctx, pgPool, batch); err != nil {
                log.Printf(\"Failed to insert batch: %v\", err)
            } else {
                processed += len(batch)
                log.Printf(\"Processed %d documents\", processed)
            }
            batch = nil
        }
    }

    // Insert remaining batch
    if len(batch) > 0 {
        if err := insertBatch(ctx, pgPool, batch); err != nil {
            log.Printf(\"Failed to insert final batch: %v\", err)
        } else {
            processed += len(batch)
        }
    }

    if err := cursor.Err(); err != nil {
        log.Fatalf(\"MongoDB cursor error: %v\", err)
    }

    log.Printf(\"Migration complete. Total documents migrated: %d\", processed)
}

func insertBatch(ctx context.Context, pool *pgxpool.Pool, batch []MigratedUser) error {
    // Use COPY for faster bulk insert (10x faster than INSERT)
    conn, err := pool.Acquire(ctx)
    if err != nil {
        return fmt.Errorf(\"failed to acquire connection: %w\", err)
    }
    defer conn.Release()

    copySQL := `
        COPY users (id, username, email, metadata, created_at, updated_at)
        FROM STDIN WITH (FORMAT CSV, HEADER FALSE)
    `
    copyConn, err := conn.Conn().PgConn().CopyFromStdin(ctx, copySQL)
    if err != nil {
        return fmt.Errorf(\"failed to start COPY: %w\", err)
    }

    for _, user := range batch {
        _, err := copyConn.Write([]byte(fmt.Sprintf(\"%s\\t%s\\t%s\\t%s\\t%s\\t%s\\n\",
            user.ID,
            user.Username,
            user.Email,
            string(user.Metadata),
            user.CreatedAt.Format(time.RFC3339),
            user.UpdatedAt.Format(time.RFC3339),
        )))
        if err != nil {
            return fmt.Errorf(\"failed to write to COPY: %w\", err)
        }
    }

    err = copyConn.Close()
    if err != nil {
        return fmt.Errorf(\"failed to close COPY: %w\", err)
    }

    return nil
}

const pgp = require('pg-promise')();
const { Client } = require('pg');
const { MongoClient } = require('mongodb');

// Configuration
const pgConfig = {
  host: 'localhost',
  port: 5432,
  database: 'vector_db',
  user: 'postgres',
  password: 'postgres',
};

const mongoConfig = {
  uri: 'mongodb://localhost:27017',
  database: 'vector_db',
  collection: 'embeddings',
};

// Initialize PostgreSQL with pgvector extension
async function initPostgreSQL() {
  const client = new Client(pgConfig);
  try {
    await client.connect();
    // Enable pgvector extension
    await client.query('CREATE EXTENSION IF NOT EXISTS vector;');
    // Create table with vector column
    await client.query(`
      CREATE TABLE IF NOT EXISTS document_embeddings (
        id SERIAL PRIMARY KEY,
        content TEXT NOT NULL,
        embedding VECTOR(384) NOT NULL,
        metadata JSONB NOT NULL,
        created_at TIMESTAMPTZ DEFAULT NOW()
      );
      CREATE INDEX IF NOT EXISTS idx_embeddings_vector ON document_embeddings 
      USING ivfflat (embedding vector_cosine_ops) WITH (lists = 100);
    `);
    console.log('PostgreSQL 18 initialized with pgvector');
    return client;
  } catch (err) {
    console.error('Failed to initialize PostgreSQL:', err);
    throw err;
  }
}

// Initialize MongoDB 8
async function initMongoDB() {
  const client = new MongoClient(mongoConfig.uri);
  try {
    await client.connect();
    await client.db(mongoConfig.database).command({ ping: 1 });
    console.log('MongoDB 8 connected successfully');
    return client;
  } catch (err) {
    console.error('Failed to connect to MongoDB:', err);
    throw err;
  }
}

// Insert embedding into PostgreSQL
async function insertPostgreSQLEmbedding(db, content, embedding, metadata) {
  try {
    const query = `
      INSERT INTO document_embeddings (content, embedding, metadata)
      VALUES ($1, $2, $3)
      RETURNING id;
    `;
    const result = await db.one(query, [
      content,
      `[${embedding.join(',')}]`, // Convert array to vector string
      metadata,
    ]);
    return result.id;
  } catch (err) {
    console.error('PostgreSQL insert failed:', err);
    throw err;
  }
}

// Search embeddings in PostgreSQL using cosine similarity
async function searchPostgreSQL(db, queryEmbedding, limit = 5) {
  try {
    const query = `
      SELECT id, content, metadata,
        1 - (embedding <=> $1) AS cosine_similarity
      FROM document_embeddings
      ORDER BY embedding <=> $1
      LIMIT $2;
    `;
    const result = await db.any(query, [
      `[${queryEmbedding.join(',')}]`,
      limit,
    ]);
    return result;
  } catch (err) {
    console.error('PostgreSQL search failed:', err);
    throw err;
  }
}

// Insert embedding into MongoDB (legacy comparison)
async function insertMongoEmbedding(collection, content, embedding, metadata) {
  try {
    const result = await collection.insertOne({
      content,
      embedding, // MongoDB stores arrays natively
      metadata,
      created_at: new Date(),
    });
    return result.insertedId;
  } catch (err) {
    console.error('MongoDB insert failed:', err);
    throw err;
  }
}

// Search embeddings in MongoDB using $vectorSearch (MongoDB 8 feature)
async function searchMongoDB(collection, queryEmbedding, limit = 5) {
  try {
    const pipeline = [
      {
        $vectorSearch: {
          index: 'embedding_index',
          path: 'embedding',
          queryVector: queryEmbedding,
          numCandidates: 100,
          limit: limit,
        },
      },
      {
        $project: {
          id: '$_id',
          content: 1,
          metadata: 1,
          cosine_similarity: { $meta: 'vectorSearchScore' },
        },
      },
    ];
    const result = await collection.aggregate(pipeline).toArray();
    return result;
  } catch (err) {
    console.error('MongoDB search failed:', err);
    throw err;
  }
}

// Main execution
async function main() {
  let pgClient, mongoClient;
  try {
    // Initialize databases
    const pg = pgp(pgConfig);
    pgClient = await initPostgreSQL();
    mongoClient = await initMongoDB();
    const mongoCollection = mongoClient.db(mongoConfig.database).collection(mongoConfig.collection);

    // Sample embeddings (384-dimensional, dummy data)
    const sampleEmbedding = Array(384).fill(0).map(() => Math.random() - 0.5);

    // Insert test data into both databases
    console.log('Inserting test data...');
    const pgId = await insertPostgreSQLEmbedding(
      pg,
      'PostgreSQL 18 vector search example',
      sampleEmbedding,
      { source: 'benchmark', tags: ['vector', 'postgres'] }
    );
    console.log(`Inserted into PostgreSQL with ID: ${pgId}`);

    const mongoId = await insertMongoEmbedding(
      mongoCollection,
      'MongoDB 8 vector search example',
      sampleEmbedding,
      { source: 'benchmark', tags: ['vector', 'mongo'] }
    );
    console.log(`Inserted into MongoDB with ID: ${mongoId}`);

    // Run search on both
    console.log('\\nSearching PostgreSQL...');
    const pgResults = await searchPostgreSQL(pg, sampleEmbedding, 1);
    console.log('PostgreSQL Results:', pgResults);

    console.log('\\nSearching MongoDB...');
    const mongoResults = await searchMongoDB(mongoCollection, sampleEmbedding, 1);
    console.log('MongoDB Results:', mongoResults);

  } catch (err) {
    console.error('Main execution failed:', err);
  } finally {
    // Cleanup
    if (pgClient) await pgClient.end();
    if (mongoClient) await mongoClient.close();
    pgp.end();
  }
}

// Run if this is the main module
if (require.main === module) {
  main();
}

-- Create a users table with JSONB and a GIN index
CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    doc JSONB NOT NULL,
    CHECK (doc->>'email' IS NOT NULL) -- Enforce non-null email
);

-- Create a GIN index for fast JSONB queries
CREATE INDEX idx_users_doc ON users USING GIN (doc);

-- Query for users with a specific role (equivalent to MongoDB's { \"doc.role\": \"admin\" })
SELECT id, doc->>'username' AS username FROM users WHERE doc->>'role' = 'admin';

-- Replace a 5-stage MongoDB aggregation pipeline with a CTE
WITH filtered_events AS (
    SELECT user_id, event_type, timestamp
    FROM events
    WHERE timestamp >= NOW() - INTERVAL '30 days'
),
user_activity AS (
    SELECT user_id,
           COUNT(*) AS total_events,
           MAX(timestamp) AS last_seen
    FROM filtered_events
    GROUP BY user_id
),
retention AS (
    SELECT user_id,
           total_events,
           last_seen,
           LAG(last_seen) OVER (PARTITION BY user_id ORDER BY last_seen) AS previous_seen
    FROM user_activity
)
SELECT user_id, total_events, last_seen
FROM retention
WHERE previous_seen IS NULL OR last_seen - previous_seen > INTERVAL '7 days';

-- Enable pgvector extension
CREATE EXTENSION IF NOT EXISTS vector;

-- Create a table with a vector column
CREATE TABLE product_embeddings (
    id SERIAL PRIMARY KEY,
    product_id INT NOT NULL,
    embedding VECTOR(384) NOT NULL,
    in_stock BOOLEAN DEFAULT TRUE
);

-- Create an HNSW index for fast vector search
CREATE INDEX idx_product_embeddings_hnsw ON product_embeddings
USING hnsw (embedding vector_cosine_ops);

-- Search for similar products that are in stock
SELECT product_id, 1 - (embedding <=> '[0.1, 0.2, ...]'::vector) AS similarity
FROM product_embeddings
WHERE in_stock = TRUE
ORDER BY embedding <=> '[0.1, 0.2, ...]'::vector
LIMIT 5;