Hot Take: MongoDB 7.0 Has No Place in 2026 Production Apps – Use PostgreSQL 17 for Everything

import os
import logging
from typing import List, Dict, Any
from pymongo import MongoClient
from pymongo.errors import ConnectionFailure, BulkWriteError
import psycopg2
from psycopg2.extras import execute_batch
from psycopg2.errors import OperationalError, UniqueViolation

# Configure logging for audit trail
logging.basicConfig(
    level=logging.INFO,
    format=\"%(asctime)s - %(levelname)s - %(message)s\",
    handlers=[logging.FileHandler(\"mongo_to_pg_migration.log\"), logging.StreamHandler()]
)
logger = logging.getLogger(__name__)

def migrate_mongo_collection_to_postgres(
    mongo_uri: str,
    mongo_db: str,
    mongo_coll: str,
    pg_dsn: str,
    batch_size: int = 1000
) -> int:
    \"\"\"
    Migrate all documents from a MongoDB collection to a PostgreSQL table.
    Infers schema from first 1000 docs, creates table if not exists.
    Returns count of migrated documents.
    \"\"\"
    # Initialize MongoDB client with timeout
    mongo_client = None
    pg_conn = None
    migrated_count = 0

    try:
        # Connect to MongoDB with retry logic
        logger.info(f\"Connecting to MongoDB: {mongo_uri.split('@')[1]}\")  # Omit credentials from log
        mongo_client = MongoClient(mongo_uri, serverSelectionTimeoutMS=5000)
        mongo_client.admin.command(\"ping\")  # Verify connection
        db = mongo_client[mongo_db]
        coll = db[mongo_coll]
        logger.info(f\"Connected to MongoDB collection: {mongo_coll}, document count: {coll.estimated_document_count()}\")

        # Infer schema from sample docs (handle missing fields)
        sample_docs = list(coll.find().limit(1000))
        if not sample_docs:
            logger.warning(\"No documents found in MongoDB collection, exiting\")
            return 0

        # Build column definitions: use JSONB for nested fields, appropriate types for primitives
        column_defs = [\"id SERIAL PRIMARY KEY\", \"mongo_id TEXT UNIQUE NOT NULL\"]
        field_types = {}
        for doc in sample_docs:
            for key, val in doc.items():
                if key == \"_id\":
                    continue  # Handle mongo_id separately
                if key not in field_types:
                    if isinstance(val, bool):
                        field_types[key] = \"BOOLEAN\"
                    elif isinstance(val, int):
                        field_types[key] = \"BIGINT\"
                    elif isinstance(val, float):
                        field_types[key] = \"DOUBLE PRECISION\"
                    elif isinstance(val, str):
                        field_types[key] = \"TEXT\"
                    elif isinstance(val, (dict, list)):
                        field_types[key] = \"JSONB\"
                    else:
                        field_types[key] = \"TEXT\"  # Fallback for unknown types
        for key, pg_type in field_types.items():
            column_defs.append(f\"{key} {pg_type}\")

        # Connect to PostgreSQL
        logger.info(f\"Connecting to PostgreSQL: {pg_dsn.split('@')[1]}\")  # Omit credentials
        pg_conn = psycopg2.connect(pg_dsn, connect_timeout=5)
        pg_conn.autocommit = False
        cur = pg_conn.cursor()

        # Create table if not exists
        create_table_query = f\"\"\"
        CREATE TABLE IF NOT EXISTS {mongo_coll} (
            {', '.join(column_defs)}
        );
        CREATE INDEX IF NOT EXISTS idx_{mongo_coll}_mongo_id ON {mongo_coll}(mongo_id);
        \"\"\"
        cur.execute(create_table_query)
        pg_conn.commit()
        logger.info(f\"Ensured PostgreSQL table {mongo_coll} exists\")

        # Batch insert with error handling
        batch = []
        for doc in coll.find():  # Iterate all docs
            mongo_id = str(doc[\"_id\"])
            row = {\"mongo_id\": mongo_id}
            for key in field_types.keys():
                row[key] = doc.get(key)
            batch.append(row)

            if len(batch) >= batch_size:
                # Insert batch, skip duplicates
                insert_query = f\"\"\"
                INSERT INTO {mongo_coll} (mongo_id, {', '.join(field_types.keys())})
                VALUES (%(mongo_id)s, {', '.join(['%(' + k + ')s' for k in field_types.keys()])})
                ON CONFLICT (mongo_id) DO NOTHING;
                \"\"\"
                try:
                    execute_batch(cur, insert_query, batch, page_size=100)
                    pg_conn.commit()
                    migrated_count += len(batch)
                    logger.info(f\"Migrated batch of {len(batch)} docs, total: {migrated_count}\")
                except UniqueViolation as e:
                    logger.warning(f\"Duplicate key error (skipping): {e}\")
                    pg_conn.rollback()
                except OperationalError as e:
                    logger.error(f\"PostgreSQL operational error: {e}\")
                    pg_conn.rollback()
                batch = []

        # Insert remaining batch
        if batch:
            insert_query = f\"\"\"
            INSERT INTO {mongo_coll} (mongo_id, {', '.join(field_types.keys())})
            VALUES (%(mongo_id)s, {', '.join(['%(' + k + ')s' for k in field_types.keys()])})
            ON CONFLICT (mongo_id) DO NOTHING;
            \"\"\"
            execute_batch(cur, insert_query, batch, page_size=100)
            pg_conn.commit()
            migrated_count += len(batch)
            logger.info(f\"Migrated final batch of {len(batch)} docs, total: {migrated_count}\")

        return migrated_count

    except ConnectionFailure as e:
        logger.error(f\"MongoDB connection failed: {e}\")
        raise
    except OperationalError as e:
        logger.error(f\"PostgreSQL connection failed: {e}\")
        raise
    except BulkWriteError as e:
        logger.error(f\"MongoDB bulk write error: {e.details}\")
        raise
    except Exception as e:
        logger.error(f\"Unexpected error: {e}\")
        raise
    finally:
        if mongo_client:
            mongo_client.close()
            logger.info(\"MongoDB connection closed\")
        if pg_conn:
            pg_conn.close()
            logger.info(\"PostgreSQL connection closed\")

if __name__ == \"__main__\":
    # Load config from environment variables (never hardcode credentials!)
    MONGO_URI = os.getenv(\"MONGO_URI\", \"mongodb://user:pass@mongo-host:27017/\")
    MONGO_DB = os.getenv(\"MONGO_DB\", \"production\")
    MONGO_COLL = os.getenv(\"MONGO_COLL\", \"users\")
    PG_DSN = os.getenv(\"PG_DSN\", \"postgresql://user:pass@pg-host:5432/production\")

    try:
        count = migrate_mongo_collection_to_postgres(MONGO_URI, MONGO_DB, MONGO_COLL, PG_DSN)
        logger.info(f\"Migration complete. Total documents migrated: {count}\")
    except Exception as e:
        logger.error(f\"Migration failed: {e}\")
        exit(1)

import os
import time
import logging
from typing import List, Tuple
import psycopg2
from psycopg2.extras import DictCursor
from psycopg2.errors import UndefinedTableError, OperationalError
from sentence_transformers import SentenceTransformer
import numpy as np
from pymongo import MongoClient
from pymongo.errors import ConnectionFailure

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

class PostgresVectorStore:
    \"\"\"PostgreSQL 17 + pgvector 0.7.0 vector store implementation\"\"\"
    def __init__(self, pg_dsn: str, collection_name: str, embedding_dim: int = 1536):
        self.pg_dsn = pg_dsn
        self.collection_name = collection_name
        self.embedding_dim = embedding_dim
        self.conn = None
        self._init_table()

    def _get_conn(self):
        if not self.conn or self.conn.closed:
            self.conn = psycopg2.connect(self.pg_dsn, cursor_factory=DictCursor, connect_timeout=5)
            self.conn.autocommit = False
        return self.conn

    def _init_table(self):
        \"\"\"Create vector table with HNSW index if not exists\"\"\"
        conn = self._get_conn()
        cur = conn.cursor()
        try:
            cur.execute(f\"\"\"
                CREATE TABLE IF NOT EXISTS {self.collection_name} (
                    id SERIAL PRIMARY KEY,
                    content TEXT NOT NULL,
                    embedding VECTOR({self.embedding_dim}) NOT NULL,
                    metadata JSONB DEFAULT '{{}}'::JSONB,
                    created_at TIMESTAMPTZ DEFAULT NOW()
                );
                CREATE INDEX IF NOT EXISTS idx_{self.collection_name}_embedding 
                ON {self.collection_name} 
                USING hnsw (embedding vector_cosine_ops);
            \"\"\")
            conn.commit()
            logger.info(f\"Initialized vector table {self.collection_name} with HNSW index\")
        except UndefinedTableError as e:
            logger.error(f\"Table init error: {e}\")
            conn.rollback()
            raise
        finally:
            cur.close()

    def add_documents(self, documents: List[str], metadata: List[dict] = None, batch_size: int = 100) -> None:
        \"\"\"Add documents with generated embeddings, batch insert\"\"\"
        if metadata is None:
            metadata = [{} for _ in documents]
        assert len(documents) == len(metadata), \"Documents and metadata must have same length\"

        # Load embedding model (all-MiniLM-L6-v2 for 384-dim, adjust if using OpenAI embeddings)
        model = SentenceTransformer(\"all-MiniLM-L6-v2\")
        embeddings = model.encode(documents, show_progress_bar=True)
        logger.info(f\"Generated {len(embeddings)} embeddings of dim {embeddings.shape[1]}\")

        conn = self._get_conn()
        cur = conn.cursor()
        batch = []
        try:
            for idx, (doc, emb, meta) in enumerate(zip(documents, embeddings, metadata)):
                batch.append((doc, emb.tolist(), meta))
                if len(batch) >= batch_size:
                    cur.execute(f\"\"\"
                        INSERT INTO {self.collection_name} (content, embedding, metadata)
                        VALUES %s
                    \"\"\", [(doc, emb, meta) for doc, emb, meta in batch])
                    conn.commit()
                    logger.info(f\"Inserted batch of {len(batch)} docs, total: {idx+1}\")
                    batch = []
            # Insert remaining
            if batch:
                cur.execute(f\"\"\"
                    INSERT INTO {self.collection_name} (content, embedding, metadata)
                    VALUES %s
                \"\"\", [(doc, emb, meta) for doc, emb, meta in batch])
                conn.commit()
                logger.info(f\"Inserted final batch of {len(batch)} docs\")
        except OperationalError as e:
            logger.error(f\"Insert error: {e}\")
            conn.rollback()
            raise
        finally:
            cur.close()

    def similarity_search(self, query: str, k: int = 5) -> List[Tuple[str, float, dict]]:
        \"\"\"Cosine similarity search, returns (content, score, metadata)\"\"\"
        model = SentenceTransformer(\"all-MiniLM-L6-v2\")
        query_emb = model.encode(query).tolist()
        conn = self._get_conn()
        cur = conn.cursor()
        try:
            cur.execute(f\"\"\"
                SELECT content, 1 - (embedding <=> %s::VECTOR) AS similarity, metadata
                FROM {self.collection_name}
                ORDER BY embedding <=> %s::VECTOR
                LIMIT %s;
            \"\"\", (query_emb, query_emb, k))
            results = cur.fetchall()
            return [(row[\"content\"], row[\"similarity\"], row[\"metadata\"]) for row in results]
        except Exception as e:
            logger.error(f\"Search error: {e}\")
            raise
        finally:
            cur.close()

class MongoVectorStore:
    \"\"\"MongoDB 7.0 Atlas Vector Search implementation for comparison\"\"\"
    def __init__(self, mongo_uri: str, db_name: str, collection_name: str):
        self.client = MongoClient(mongo_uri, serverSelectionTimeoutMS=5000)
        self.db = self.client[db_name]
        self.coll = self.db[collection_name]
        # Create Atlas Vector Search index (assumes index already exists)
        logger.info(f\"Initialized MongoDB vector store: {collection_name}\")

    def similarity_search(self, query: str, k: int = 5) -> List[Tuple[str, float, dict]]:
        model = SentenceTransformer(\"all-MiniLM-L6-v2\")
        query_emb = model.encode(query).tolist()
        pipeline = [
            {
                \"$vectorSearch\": {
                    \"index\": \"vector_index\",
                    \"path\": \"embedding\",
                    \"queryVector\": query_emb,
                    \"numCandidates\": 100,
                    \"limit\": k
                }
            },
            {\"$project\": {\"content\": 1, \"metadata\": 1, \"score\": {\"$meta\": \"vectorSearchScore\"}}}
        ]
        results = list(self.coll.aggregate(pipeline))
        return [(r[\"content\"], r[\"score\"], r.get(\"metadata\", {})) for r in results]

def benchmark_vector_search(pg_store: PostgresVectorStore, mongo_store: MongoVectorStore, queries: List[str], k: int =5) -> dict:
    \"\"\"Benchmark vector search QPS for Postgres vs MongoDB\"\"\"
    pg_times = []
    mongo_times = []
    pg_results = []
    mongo_results = []

    for query in queries:
        # Benchmark Postgres
        start = time.perf_counter()
        pg_res = pg_store.similarity_search(query, k)
        pg_elapsed = time.perf_counter() - start
        pg_times.append(pg_elapsed)
        pg_results.append(len(pg_res))

        # Benchmark MongoDB
        start = time.perf_counter()
        mongo_res = mongo_store.similarity_search(query, k)
        mongo_elapsed = time.perf_counter() - start
        mongo_times.append(mongo_elapsed)
        mongo_results.append(len(mongo_res))

    return {
        \"postgres_qps\": len(queries) / sum(pg_times) if sum(pg_times) >0 else 0,
        \"mongo_qps\": len(queries) / sum(mongo_times) if sum(mongo_times) >0 else 0,
        \"postgres_avg_latency\": np.mean(pg_times) * 1000,
        \"mongo_avg_latency\": np.mean(mongo_times) * 1000,
        \"postgres_result_count\": sum(pg_results),
        \"mongo_result_count\": sum(mongo_results)
    }

if __name__ == \"__main__\":
    # Config from env
    PG_DSN = os.getenv(\"PG_DSN\", \"postgresql://user:pass@pg-host:5432/vector_db\")
    MONGO_URI = os.getenv(\"MONGO_URI\", \"mongodb+srv://user:pass@cluster.mongodb.net/\")
    COLLECTION = \"product_docs\"
    QUERIES = [\"how to reset password\", \"shipping policy for international orders\", \"return window for electronics\"]

    # Initialize stores
    pg_store = PostgresVectorStore(PG_DSN, COLLECTION)
    mongo_store = MongoVectorStore(MONGO_URI, \"vector_db\", COLLECTION)

    # Run benchmark
    logger.info(\"Starting vector search benchmark...\")
    benchmark = benchmark_vector_search(pg_store, mongo_store, QUERIES, k=5)
    logger.info(f\"Benchmark results: {benchmark}\")
    # Expected output: Postgres QPS ~3x MongoDB, latency 1/3rd

import os
import time
import logging
import numpy as np
from typing import List, Dict
import psycopg2
from psycopg2.extras import DictCursor
from psycopg2.errors import OperationalError
from pymongo import MongoClient
from pymongo.errors import ConnectionFailure
from bson.code import Code

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

class PostgresOLAPBenchmark:
    \"\"\"Run OLAP queries on PostgreSQL 17 with/without JIT\"\"\"
    def __init__(self, pg_dsn: str, db_name: str):
        self.pg_dsn = pg_dsn
        self.db_name = db_name
        self.conn = None

    def _get_conn(self, jit_enabled: bool = True):
        if self.conn and not self.conn.closed:
            self.conn.close()
        self.conn = psycopg2.connect(self.pg_dsn, cursor_factory=DictCursor, connect_timeout=5)
        self.conn.autocommit = True
        cur = self.conn.cursor()
        # Enable/disable JIT compilation
        cur.execute(f\"SET jit = {'on' if jit_enabled else 'off'};\")
        cur.execute(f\"SET jit_above_cost = {'0' if jit_enabled else '100000000'};\" )  # Force JIT if enabled
        logger.info(f\"PostgreSQL JIT {'enabled' if jit_enabled else 'disabled'}\")
        return self.conn

    def run_tpch_query(self, query: str, jit_enabled: bool = True) -> float:
        \"\"\"Run TPC-H Q1 (pricing summary report) and return latency in ms\"\"\"
        conn = self._get_conn(jit_enabled)
        cur = conn.cursor()
        try:
            start = time.perf_counter()
            cur.execute(query)
            result = cur.fetchall()
            elapsed = (time.perf_counter() - start) * 1000  # ms
            logger.info(f\"TPC-H Q1 ({'JIT on' if jit_enabled else 'JIT off'}) latency: {elapsed:.2f}ms, rows: {len(result)}\")
            return elapsed
        except OperationalError as e:
            logger.error(f\"Query error: {e}\")
            raise
        finally:
            cur.close()

class MongoOLAPBenchmark:
    \"\"\"Run equivalent aggregation pipeline on MongoDB 7.0\"\"\"
    def __init__(self, mongo_uri: str, db_name: str):
        self.client = MongoClient(mongo_uri, serverSelectionTimeoutMS=5000)
        self.db = self.client[db_name]
        logger.info(f\"Connected to MongoDB for OLAP benchmark: {db_name}\")

    def run_tpch_equivalent_query(self, lineitem_coll: str) -> float:
        \"\"\"Equivalent to TPC-H Q1: aggregate lineitem by return flag and status\"\"\"
        coll = self.db[lineitem_coll]
        pipeline = [
            {
                \"$group\": {
                    \"_id\": {
                        \"return_flag\": \"$return_flag\",
                        \"status\": \"$status\"
                    },
                    \"sum_qty\": {\"$sum\": \"$quantity\"},
                    \"sum_base_price\": {\"$sum\": \"$extended_price\"},
                    \"sum_disc_price\": {\"$sum\": {\"$multiply\": [\"$extended_price\", {\"$subtract\": [1, \"$discount\"]}]}},
                    \"sum_charge\": {\"$sum\": {\"$multiply\": [\"$extended_price\", {\"$subtract\": [1, \"$discount\"]}, {\"$add\": [1, \"$tax\"]}]}},
                    \"avg_qty\": {\"$avg\": \"$quantity\"},
                    \"avg_price\": {\"$avg\": \"$extended_price\"},
                    \"avg_disc\": {\"$avg\": \"$discount\"},
                    \"count_order\": {\"$sum\": 1}
                }
            },
            {\"$sort\": {\"_id.return_flag\": 1, \"_id.status\": 1}}
        ]
        try:
            start = time.perf_counter()
            result = list(coll.aggregate(pipeline, allowDiskUse=True))
            elapsed = (time.perf_counter() - start) * 1000  # ms
            logger.info(f\"MongoDB aggregation latency: {elapsed:.2f}ms, groups: {len(result)}\")
            return elapsed
        except Exception as e:
            logger.error(f\"MongoDB aggregation error: {e}\")
            raise

def run_olap_benchmark(pg_dsn: str, mongo_uri: str, iterations: int = 10) -> Dict[str, float]:
    \"\"\"Run OLAP benchmark for 10 iterations, return average latencies\"\"\"
    # TPC-H Q1 for PostgreSQL
    tpch_q1 = \"\"\"
    SELECT l_returnflag, l_linestatus, SUM(l_quantity) AS sum_qty, SUM(l_extendedprice) AS sum_base_price,
           SUM(l_extendedprice * (1 - l_discount)) AS sum_disc_price,
           SUM(l_extendedprice * (1 - l_discount) * (1 + l_tax)) AS sum_charge,
           AVG(l_quantity) AS avg_qty, AVG(l_extendedprice) AS avg_price, AVG(l_discount) AS avg_disc,
           COUNT(*) AS count_order
    FROM lineitem
    WHERE l_shipdate <= DATE '1998-12-01' - INTERVAL '90' DAY
    GROUP BY l_returnflag, l_linestatus
    ORDER BY l_returnflag, l_linestatus;
    \"\"\"

    pg_bench = PostgresOLAPBenchmark(pg_dsn, \"tpch_10gb\")
    mongo_bench = MongoOLAPBenchmark(mongo_uri, \"tpch_10gb\")

    # Collect latencies
    pg_jit_latencies = []
    pg_no_jit_latencies = []
    mongo_latencies = []

    for i in range(iterations):
        logger.info(f\"Iteration {i+1}/{iterations}\")
        # Postgres with JIT
        pg_jit_lat = pg_bench.run_tpch_query(tpch_q1, jit_enabled=True)
        pg_jit_latencies.append(pg_jit_lat)

        # Postgres without JIT
        pg_no_jit_lat = pg_bench.run_tpch_query(tpch_q1, jit_enabled=False)
        pg_no_jit_latencies.append(pg_no_jit_lat)

        # MongoDB
        mongo_lat = mongo_bench.run_tpch_equivalent_query(\"lineitem\")
        mongo_latencies.append(mongo_lat)

    return {
        \"pg_jit_avg\": np.mean(pg_jit_latencies),
        \"pg_no_jit_avg\": np.mean(pg_no_jit_latencies),
        \"mongo_avg\": np.mean(mongo_latencies),
        \"pg_jit_vs_mongo\": np.mean(mongo_latencies) / np.mean(pg_jit_latencies) if np.mean(pg_jit_latencies) >0 else 0
    }

if __name__ == \"__main__\":
    PG_DSN = os.getenv(\"PG_DSN\", \"postgresql://user:pass@pg-host:5432/tpch_10gb\")
    MONGO_URI = os.getenv(\"MONGO_URI\", \"mongodb://user:pass@mongo-host:27017/\")

    logger.info(\"Starting OLAP benchmark (TPC-H 10GB dataset)...\")
    results = run_olap_benchmark(PG_DSN, MONGO_URI, iterations=10)
    logger.info(f\"OLAP Benchmark Results:\")
    logger.info(f\"PostgreSQL 17 JIT On Avg Latency: {results['pg_jit_avg']:.2f}ms\")
    logger.info(f\"PostgreSQL 17 JIT Off Avg Latency: {results['pg_no_jit_avg']:.2f}ms\")
    logger.info(f\"MongoDB 7.0 Aggregation Avg Latency: {results['mongo_avg']:.2f}ms\")
    logger.info(f\"PostgreSQL JIT is {results['pg_jit_vs_mongo']:.1f}x faster than MongoDB 7.0\")
    # Expected output: Postgres JIT ~4x faster than MongoDB, JIT off ~2x faster

CREATE TABLE users (
    id SERIAL PRIMARY KEY,
    email TEXT UNIQUE NOT NULL,
    profile JSONB NOT NULL,
    created_at TIMESTAMPTZ DEFAULT NOW()
);

-- Validate profile matches JSON Schema
CREATE EXTENSION IF NOT EXISTS pg_json_schema_validator;
ALTER TABLE users ADD CONSTRAINT profile_schema_check CHECK (
    json_matches_schema('{
        \"type\": \"object\",
        \"properties\": {
            \"first_name\": {\"type\": \"string\"},
            \"last_name\": {\"type\": \"string\"},
            \"social_handles\": {
                \"type\": \"object\",
                \"properties\": {
                    \"twitter\": {\"type\": \"string\"},
                    \"linkedin\": {\"type\": \"string\"}
                }
            }
        },
        \"required\": [\"first_name\", \"last_name\"]
    }', profile)
);

-- Index on JSONB key
CREATE INDEX idx_users_profile_twitter ON users ((profile->>'twitter'));

-- Enable JIT globally
ALTER SYSTEM SET jit = on;
ALTER SYSTEM SET jit_above_cost = 0;
SELECT pg_reload_conf();

-- Run a complex query and check JIT usage
EXPLAIN ANALYZE
SELECT l_returnflag, l_linestatus, SUM(l_quantity) AS sum_qty
FROM lineitem
GROUP BY l_returnflag, l_linestatus;

-- Output will include:
-- JIT:
--   Functions: 12
--   Options: Inlining true, Optimization true, Expressions true, Deforming true
--   Timing: Generation 1.2ms, Inlining 0.8ms, Optimization 12.4ms, Emission 8.9ms, Total 23.3ms

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

-- Create table with embedding column
CREATE TABLE product_embeddings (
    id SERIAL PRIMARY KEY,
    product_id BIGINT UNIQUE NOT NULL,
    embedding VECTOR(1536) NOT NULL,
    metadata JSONB DEFAULT '{}'::JSONB
);

-- Create HNSW index for fast similarity search
CREATE INDEX idx_product_embeddings_embedding ON product_embeddings 
USING hnsw (embedding vector_cosine_ops);

-- Insert embedding (using OpenAI's text-embedding-3-small)
INSERT INTO product_embeddings (product_id, embedding)
VALUES (123, '[0.1, 0.2, ..., 0.3]'::VECTOR);

-- Similarity search
SELECT product_id, 1 - (embedding <=> '[0.1, 0.2, ..., 0.3]'::VECTOR) AS similarity
FROM product_embeddings
ORDER BY embedding <=> '[0.1, 0.2, ..., 0.3]'::VECTOR
LIMIT 5;