PostgreSQL to Billions: The Architecture Migration Nobody Talks About

1
from pydantic_settings import BaseSettings
2

3
class Settings(BaseSettings):
4
  yugabyte_host: str = "yugabyte-cluster.local"
5
  yugabyte_port: int = 5433
6
  yugabyte_db: str = "documents_db"
7
  yugabyte_user: str = "app_user"
8
  yugabyte_password: str
9

10
settings = Settings()

1
import psycopg2
2
from psycopg2.pool import ThreadedConnectionPool
3
from config import settings
4

5
# Connection pool (standard PostgreSQL pattern)
6
pg_pool = ThreadedConnectionPool(
7
  minconn=5,
8
  maxconn=20,
9
  host=settings.yugabyte_host,
10
  port=settings.yugabyte_port,
11
  database=settings.yugabyte_db,
12
  user=settings.yugabyte_user,
13
  password=settings.yugabyte_password
14
)
15

16
def get_db_connection():
17
  return pg_pool.getconn()
18

19
def release_db_connection(conn):
20
  pg_pool.putconn(conn)

1
import clickhouse_connect
2
from config import settings
3

4
class MyScaleClient:
5
  def __init__(self):
6
      self.client = clickhouse_connect.get_client(
7
          host=settings.myscale_host,
8
          port=8123,
9
          username=settings.myscale_user,
10
          password=settings.myscale_password
11
      )
12

13
  def insert_vectors(self, vectors_data):
14
      """Batch insert vectors"""
15
      self.client.insert(
16
          'document_vectors',
17
          vectors_data,
18
          column_names=['doc_id', 'title', 'category',
19
                       'embedding', 'metadata',
20
                       'created_at', 'updated_at']
21
      )
22

23
  def search_similar(self, query_vector, category=None, limit=10):
24
      """Vector similarity search with optional filtering"""
25
      where_clause = f"WHERE category = '{category}'" if category else ""
26

27
      query = f"""
28
          SELECT
29
              doc_id,
30
              title,
31
              category,
32
              distance(embedding, {query_vector}) as similarity,
33
              metadata
34
          FROM document_vectors
35
          {where_clause}
36
          ORDER BY similarity ASC
37
          LIMIT {limit}
38
      """
39

40
      return self.client.query(query).result_rows

1
import time
2
from typing import List, Dict
3
import numpy as np
4
from sentence_transformers import SentenceTransformer
5
from database import get_db_connection, release_db_connection
6
from myscale_client import MyScaleClient
7
import logging
8

9
logging.basicConfig(level=logging.INFO)
10
logger = logging.getLogger(__name__)
11

12
class SyncPipeline:
13
  def __init__(self, batch_size: int = 1000):
14
      self.batch_size = batch_size
15
      self.embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
16
      self.myscale = MyScaleClient()
17

18
  def fetch_unprocessed_documents(self, conn) -> List[Dict]:
19
      """Fetch documents that haven't been synced to MyScaleDB"""
20
      cursor = conn.cursor()
21
      cursor.execute("""
22
          SELECT id, content, title, category, metadata,
23
                 created_at, updated_at
24
          FROM documents
25
          WHERE NOT embedding_synced
26
          ORDER BY created_at
27
          LIMIT %s
28
      """, (self.batch_size,))
29

30
      columns = [desc[0] for desc in cursor.description]
31
      rows = cursor.fetchall()
32

33
      return [dict(zip(columns, row)) for row in rows]
34

35
  def compute_embeddings(self, documents: List[Dict]) -> np.ndarray:
36
      """Compute embeddings for documents"""
37
      texts = [
38
          f"{doc['title']} {doc['content']}"
39
          for doc in documents
40
      ]
41

42
      logger.info(f"Computing embeddings for {len(texts)} documents...")
43
      embeddings = self.embedding_model.encode(
44
          texts,
45
          show_progress_bar=True,
46
          batch_size=32
47
      )
48

49
      return embeddings
50

51
  def sync_to_myscale(self, documents: List[Dict], embeddings: np.ndarray):
52
      """Push vectors to MyScaleDB"""
53
      vectors_data = []
54

55
      for doc, embedding in zip(documents, embeddings):
56
          vectors_data.append({
57
              'doc_id': doc['id'],
58
              'title': doc['title'],
59
              'category': doc['category'],
60
              'embedding': embedding.tolist(),
61
              'metadata': str(doc['metadata']),
62
              'created_at': doc['created_at'],
63
              'updated_at': doc['updated_at']
64
          })
65

66
      logger.info(f"Inserting {len(vectors_data)} vectors into MyScaleDB...")
67
      self.myscale.insert_vectors(vectors_data)
68

69
  def mark_as_synced(self, conn, doc_ids: List[int]):
70
      """Mark documents as synced in YugabyteDB"""
71
      cursor = conn.cursor()
72
      cursor.execute("""
73
          UPDATE documents
74
          SET embedding_synced = TRUE,
75
              updated_at = NOW()
76
          WHERE id = ANY(%s)
77
      """, (doc_ids,))
78
      conn.commit()
79

80
      logger.info(f"Marked {len(doc_ids)} documents as synced")
81

82
  def run_batch(self) -> int:
83
      """Process one batch of documents"""
84
      conn = get_db_connection()
85

86
      try:
87
          # Fetch unprocessed documents
88
          documents = self.fetch_unprocessed_documents(conn)
89

90
          if not documents:
91
              logger.info("No documents to process")
92
              return 0
93

94
          # Compute embeddings
95
          embeddings = self.compute_embeddings(documents)
96

97
          # Sync to MyScaleDB
98
          self.sync_to_myscale(documents, embeddings)
99

100
          # Mark as synced
101
          doc_ids = [doc['id'] for doc in documents]
102
          self.mark_as_synced(conn, doc_ids)
103

104
          logger.info(f"Successfully processed {len(documents)} documents")
105
          return len(documents)
106

107
      except Exception as e:
108
          logger.error(f"Error in sync pipeline: {e}")
109
          conn.rollback()
110
          raise
111

112
      finally:
113
          release_db_connection(conn)
114

115
  def run_continuous(self, interval_seconds: int = 60):
116
      """Run pipeline continuously"""
117
      logger.info(f"Starting continuous sync (interval: {interval_seconds}s)")
118

119
      while True:
120
          try:
121
              processed = self.run_batch()
122

123
              if processed == 0:
124
                  time.sleep(interval_seconds)
125
              else:
126
                  # If we processed a full batch, check immediately
127
                  # for more (backlog catching up)
128
                  if processed < self.batch_size:
129
                      time.sleep(interval_seconds)
130

131
          except KeyboardInterrupt:
132
              logger.info("Stopping sync pipeline...")
133
              break
134
          except Exception as e:
135
              logger.error(f"Pipeline error: {e}")
136
              time.sleep(interval_seconds)
137

138
# Usage
139
if __name__ == "__main__":
140
  pipeline = SyncPipeline(batch_size=1000)
141
  pipeline.run_continuous(interval_seconds=60)

1
from fastapi import FastAPI, HTTPException
2
from pydantic import BaseModel
3
from typing import List, Optional
4
from database import get_db_connection, release_db_connection
5
from myscale_client import MyScaleClient
6
from sentence_transformers import SentenceTransformer
7

8
app = FastAPI()
9
myscale = MyScaleClient()
10
embedding_model = SentenceTransformer('all-MiniLM-L6-v2')
11

12
class DocumentCreate(BaseModel):
13
  content: str
14
  title: str
15
  category: str
16
  metadata: dict = {}
17

18
class SearchQuery(BaseModel):
19
  query: str
20
  category: Optional[str] = None
21
  limit: int = 10
22

23
@app.post("/documents/")
24
async def create_document(doc: DocumentCreate):
25
  """Create document in YugabyteDB (will be synced to MyScaleDB)"""
26
  conn = get_db_connection()
27

28
  try:
29
      cursor = conn.cursor()
30
      cursor.execute("""
31
          INSERT INTO documents (content, title, category, metadata)
32
          VALUES (%s, %s, %s, %s)
33
          RETURNING id, created_at
34
      """, (doc.content, doc.title, doc.category, doc.metadata))
35

36
      result = cursor.fetchone()
37
      conn.commit()
38

39
      return {
40
          "id": result[0],
41
          "created_at": result[1],
42
          "status": "created",
43
          "note": "Will be synced to vector DB in next pipeline run"
44
      }
45

46
  finally:
47
      release_db_connection(conn)
48

49
@app.post("/search/")
50
async def search_documents(query: SearchQuery):
51
  """Semantic search via MyScaleDB"""
52

53
  # Compute query embedding
54
  query_embedding = embedding_model.encode(query.query).tolist()
55

56
  # Search in MyScaleDB
57
  results = myscale.search_similar(
58
      query_vector=query_embedding,
59
      category=query.category,
60
      limit=query.limit
61
  )
62

63
  # Enrich with full document data from YugabyteDB
64
  doc_ids = [r[0] for r in results]
65

66
  conn = get_db_connection()
67
  try:
68
      cursor = conn.cursor()
69
      cursor.execute("""
70
          SELECT id, content, title, category, metadata
71
          FROM documents
72
          WHERE id = ANY(%s)
73
      """, (doc_ids,))
74

75
      docs_dict = {row[0]: row for row in cursor.fetchall()}
76

77
  finally:
78
      release_db_connection(conn)
79

80
  # Combine vector search results with full documents
81
  enriched_results = []
82
  for doc_id, title, category, similarity, metadata in results:
83
      if doc_id in docs_dict:
84
          full_doc = docs_dict[doc_id]
85
          enriched_results.append({
86
              "id": doc_id,
87
              "title": title,
88
              "content": full_doc[1],
89
              "category": category,
90
              "similarity": similarity,
91
              "metadata": metadata
92
          })
93

94
  return {"results": enriched_results}
95

96
@app.get("/health/sync-lag")
97
async def check_sync_lag():
98
  """Monitor sync lag between YugabyteDB and MyScaleDB"""
99
  conn = get_db_connection()
100

101
  try:
102
      cursor = conn.cursor()
103

104
      # Check unsynced count
105
      cursor.execute("""
106
          SELECT COUNT(*) as unsynced
107
          FROM documents
108
          WHERE NOT embedding_synced
109
      """)
110
      unsynced_count = cursor.fetchone()[0]
111

112
      # Check oldest unsynced
113
      cursor.execute("""
114
          SELECT MAX(created_at) as oldest_unsynced
115
          FROM documents
116
          WHERE NOT embedding_synced
117
      """)
118
      oldest_unsynced = cursor.fetchone()[0]
119

120
      return {
121
          "unsynced_documents": unsynced_count,
122
          "oldest_unsynced": oldest_unsynced,
123
          "status": "ok" if unsynced_count < 10000 else "warning"
124
      }
125

126
  finally:
127
      release_db_connection(conn)

1
import psycopg2
2
import time
3

4
# Export from PostgreSQL
5
pg_conn = psycopg2.connect("postgresql://localhost/olddb")
6
yb_conn = psycopg2.connect("postgresql://localhost:5433/newdb")
7

8
pg_cursor = pg_conn.cursor()
9
yb_cursor = yb_conn.cursor()
10

11
# Sample 10% of data
12
pg_cursor.execute("""
13
  SELECT id, content, title, category, metadata, created_at
14
  FROM documents
15
  TABLESAMPLE SYSTEM (10)
16
""")
17

18
batch = []
19
batch_size = 1000
20

21
for row in pg_cursor:
22
  batch.append(row)
23

24
  if len(batch) >= batch_size:
25
      # Bulk insert to YugabyteDB
26
      start = time.time()
27
      yb_cursor.executemany("""
28
          INSERT INTO documents (id, content, title, category, metadata, created_at)
29
          VALUES (%s, %s, %s, %s, %s, %s)
30
      """, batch)
31
      yb_conn.commit()
32
      elapsed = time.time() - start
33

34
      rate = len(batch) / elapsed
35
      print(f"Inserted {len(batch)} docs in {elapsed:.2f}s ({rate:.0f} docs/sec)")
36

37
      batch = []
38

39
print("Migration test complete")

1
from typing import Iterator, List, Tuple
2
import psycopg2
3
from psycopg2.extras import execute_values
4
import logging
5

6
logging.basicConfig(level=logging.INFO)
7
logger = logging.getLogger(__name__)
8

9
class MigrationOrchestrator:
10
  def __init__(self, pg_dsn: str, yb_dsn: str, batch_size: int = 5000):
11
      self.pg_conn = psycopg2.connect(pg_dsn)
12
      self.yb_conn = psycopg2.connect(yb_dsn)
13
      self.batch_size = batch_size
14

15
  def stream_documents(self) -> Iterator[List[Tuple]]:
16
      """Stream documents from PostgreSQL in batches"""
17
      cursor = self.pg_conn.cursor(name='migration_cursor')
18
      cursor.itersize = self.batch_size
19

20
      cursor.execute("""
21
          SELECT id, content, title, category, metadata,
22
                 created_at, updated_at
23
          FROM documents
24
          ORDER BY id
25
      """)
26

27
      while True:
28
          batch = cursor.fetchmany(self.batch_size)
29
          if not batch:
30
              break
31
          yield batch
32

33
  def migrate_batch(self, batch: List[Tuple]) -> int:
34
      """Migrate one batch to YugabyteDB"""
35
      cursor = self.yb_conn.cursor()
36

37
      execute_values(
38
          cursor,
39
          """
40
          INSERT INTO documents
41
              (id, content, title, category, metadata, created_at, updated_at)
42
          VALUES %s
43
          ON CONFLICT (id) DO UPDATE SET
44
              content = EXCLUDED.content,
45
              title = EXCLUDED.title,
46
              category = EXCLUDED.category,
47
              metadata = EXCLUDED.metadata,
48
              updated_at = EXCLUDED.updated_at
49
          """,
50
          batch
51
      )
52

53
      self.yb_conn.commit()
54
      return len(batch)
55

56
  def run(self):
57
      """Execute full migration"""
58
      total_migrated = 0
59
      start_time = time.time()
60

61
      logger.info("Starting migration...")
62

63
      for batch in self.stream_documents():
64
          migrated = self.migrate_batch(batch)
65
          total_migrated += migrated
66

67
          elapsed = time.time() - start_time
68
          rate = total_migrated / elapsed
69

70
          logger.info(
71
              f"Migrated {total_migrated:,} documents "
72
              f"({rate:.0f} docs/sec, {elapsed:.1f}s elapsed)"
73
          )
74

75
      logger.info(f"Migration complete: {total_migrated:,} documents")
76

77
      # Verify
78
      self.verify_migration(total_migrated)
79

80
  def verify_migration(self, expected_count: int):
81
      """Verify migration integrity"""
82
      cursor = self.yb_conn.cursor()
83
      cursor.execute("SELECT COUNT(*) FROM documents")
84
      actual_count = cursor.fetchone()[0]
85

86
      if actual_count == expected_count:
87
          logger.info(f"✓ Verification passed: {actual_count:,} documents")
88
      else:
89
          logger.error(
90
              f"✗ Count mismatch: expected {expected_count:,}, "
91
              f"got {actual_count:,}"
92
          )
93

94
# Run migration
95
if __name__ == "__main__":
96
  orchestrator = MigrationOrchestrator(
97
      pg_dsn="postgresql://localhost/olddb",
98
      yb_dsn="postgresql://localhost:5433/newdb",
99
      batch_size=5000
100
  )
101
  orchestrator.run()

1
import psycopg2
2
import time
3

4
def check_yugabyte_health(conn):
5
  cursor = conn.cursor()
6

7
  # Replication lag
8
  cursor.execute("""
9
      SELECT
10
          application_name,
11
          client_addr,
12
          state,
13
          sync_state,
14
          replay_lag
15
      FROM pg_stat_replication
16
  """)
17

18
  print("\n=== Replication Status ===")
19
  for row in cursor.fetchall():
20
      app, addr, state, sync, lag = row
21
      print(f"{app} ({addr}): {state}, {sync}, lag: {lag}")
22

23
  # Connection stats
24
  cursor.execute("""
25
      SELECT
26
          state,
27
          COUNT(*) as count
28
      FROM pg_stat_activity
29
      WHERE datname = current_database()
30
      GROUP BY state
31
  """)
32

33
  print("\n=== Connection States ===")
34
  for state, count in cursor.fetchall():
35
      print(f"{state}: {count}")
36

37
  # Query performance
38
  cursor.execute("""
39
      SELECT
40
          query,
41
          calls,
42
          mean_exec_time,
43
          max_exec_time
44
      FROM pg_stat_statements
45
      WHERE mean_exec_time > 100  -- Queries slower than 100ms
46
      ORDER BY mean_exec_time DESC
47
      LIMIT 5
48
  """)
49

50
  print("\n=== Slow Queries (>100ms) ===")
51
  for query, calls, mean, max_time in cursor.fetchall():
52
      print(f"Mean: {mean:.1f}ms, Max: {max_time:.1f}ms, Calls: {calls}")
53
      print(f"  {query[:80]}...")
54

55
# Run every 60s
56
while True:
57
  try:
58
      conn = psycopg2.connect("postgresql://yugabyte:5433/documents_db")
59
      check_yugabyte_health(conn)
60
      conn.close()
61
  except Exception as e:
62
      print(f"Error: {e}")
63

64
  time.sleep(60)

1
import clickhouse_connect
2

3
def check_myscale_health(client):
4
  # Query performance
5
  result = client.query("""
6
      SELECT
7
          query_duration_ms,
8
          read_rows,
9
          read_bytes,
10
          memory_usage
11
      FROM system.query_log
12
      WHERE type = 'QueryFinish'
13
        AND event_time > now() - INTERVAL 1 HOUR
14
      ORDER BY query_duration_ms DESC
15
      LIMIT 10
16
  """)
17

18
  print("\n=== Slowest Queries (Last Hour) ===")
19
  for row in result.result_rows:
20
      duration, rows, bytes_read, memory = row
21
      print(f"Duration: {duration}ms, Rows: {rows:,}, "
22
            f"Bytes: {bytes_read:,}, Memory: {memory:,}")
23

24
  # Merges and mutations
25
  result = client.query("""
26
      SELECT
27
          database,
28
          table,
29
          COUNT(*) as parts,
30
          SUM(rows) as total_rows,
31
          formatReadableSize(SUM(bytes_on_disk)) as size
32
      FROM system.parts
33
      WHERE active
34
      GROUP BY database, table
35
  """)
36

37
  print("\n=== Table Statistics ===")
38
  for db, table, parts, rows, size in result.result_rows:
39
      print(f"{db}.{table}: {parts} parts, {rows:,} rows, {size}")
40

41
  # Check for excessive parts (needs merging)
42
  if parts > 100:
43
      print(f"⚠ Warning: {table} has {parts} parts, consider OPTIMIZE")
44

45
client = clickhouse_connect.get_client(host='localhost', port=8123)
46
check_myscale_health(client)

1
import psycopg2
2
from datetime import datetime, timedelta
3

4
def check_sync_lag(yugabyte_conn):
5
  cursor = yugabyte_conn.cursor()
6

7
  # Unsynced documents count
8
  cursor.execute("""
9
      SELECT COUNT(*) FROM documents WHERE NOT embedding_synced
10
  """)
11
  unsynced = cursor.fetchone()[0]
12

13
  # Oldest unsynced document
14
  cursor.execute("""
15
      SELECT
16
          MIN(created_at) as oldest,
17
          NOW() - MIN(created_at) as lag
18
      FROM documents
19
      WHERE NOT embedding_synced
20
  """)
21

22
  oldest, lag = cursor.fetchone()
23

24
  print("\n=== Sync Pipeline Status ===")
25
  print(f"Unsynced documents: {unsynced:,}")
26

27
  if oldest:
28
      print(f"Oldest unsynced: {oldest}")
29
      print(f"Lag: {lag}")
30

31
      # Alert if lag > 10 minutes
32
      if lag > timedelta(minutes=10):
33
          print("🚨 ALERT: Sync lag exceeds 10 minutes!")
34
          return False
35

36
  return True
37

38
# Integration with alerting
39
conn = psycopg2.connect("postgresql://yugabyte:5433/documents_db")
40
healthy = check_sync_lag(conn)
41

42
if not healthy:
43
  # Send alert (PagerDuty, Slack, etc.)
44
  send_alert("Sync pipeline lagging behind")