Final project — phase 2: partitioning, advisory lock, recursive categories

You continue the integrating project. This capsule covers the 3 remaining components: partitioning of comments with a zero-downtime migration, recursive categories with a CTE, and an advisory lock to protect the refresh and re-indexing crons.

When you finish, all 6 components are implemented. Capsule 08 closes with the final documentation.

---

Component 4: partitioning of comments by month

Problem: the comments table has 1M+ rows and grows without stopping. "Recent comments" queries get slow. Solution: partitioning by month (module 4).

Zero-downtime migration

This migration requires zero-downtime because the comments table is active. Pattern (from guide #13 module 5 — expand-contract):

Step 1: create the new partitioned table (empty)

# alembic/versions/XXX_partition_comments_step1.py
def upgrade() -> None:
    # Create the new partitioned table
    op.execute("""
        CREATE TABLE comments_partitioned (
            id SERIAL,
            post_id INTEGER NOT NULL,
            user_id INTEGER NOT NULL,
            body TEXT NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
            PRIMARY KEY (id, created_at)  -- composite due to the partitioning requirement
        ) PARTITION BY RANGE (created_at)
    """)

    # Create partitions for the last 12 months + 3 future months
    for i in range(-12, 4):
        # Monthly partitions
        # ... logic to generate partitions
        pass

    # Indexes on the main table (applied to partitions)
    op.execute("CREATE INDEX idx_comments_part_post ON comments_partitioned (post_id, created_at DESC)")

Step 2: copy data in batches

# scripts/migrate_comments.py
async def migrate_in_batches():
    BATCH = 10_000
    async with SessionLocal() as session:
        last_id = 0
        while True:
            result = await session.execute(text("""
                INSERT INTO comments_partitioned (id, post_id, user_id, body, created_at)
                SELECT id, post_id, user_id, body, created_at
                FROM comments
                WHERE id > :last_id
                ORDER BY id
                LIMIT :batch
                RETURNING id
            """), {"last_id": last_id, "batch": BATCH})

            rows = result.all()
            if not rows:
                break
            last_id = rows[-1][0]
            await session.commit()
            print(f"Migrated up to id {last_id}")

Step 3: deploy the app that writes to both tables

# app/services/comments.py
async def create_comment(session, data):
    # Double write during the transition
    comment = Comment(**data)
    session.add(comment)
    await session.flush()

    # Also to the partitioned table
    await session.execute(text("""
        INSERT INTO comments_partitioned (id, post_id, user_id, body, created_at)
        VALUES (:id, :post_id, :user_id, :body, :created_at)
    """), {**data, "id": comment.id})

    await session.commit()
    return comment

Step 4: switch reads to partitioned

# Reads now from partitioned
async def get_recent_comments(session, post_id):
    result = await session.execute(text("""
        SELECT * FROM comments_partitioned
        WHERE post_id = :pid
        ORDER BY created_at DESC LIMIT 50
    """), {"pid": post_id})
    return result.mappings().all()

Step 5: stop the dual-write, drop the old table

def upgrade() -> None:
    op.execute("DROP TABLE comments")
    op.execute("ALTER TABLE comments_partitioned RENAME TO comments")

This requires careful coordination. For details, guide #13 module 5.

Benchmark with partitioning

EXPLAIN ANALYZE
SELECT * FROM comments
WHERE post_id = 123
  AND created_at > NOW() - INTERVAL '7 days'
ORDER BY created_at DESC LIMIT 50;

Append (cost=...)
  -> Index Scan on comments_2026_05  (only partition scanned)
        Index Cond: (post_id = 123)

Partition pruning: only the current month is scanned. ~10x faster than the full table.

---

Component 5: recursive categories with a CTE

Case: show the category subtree with a post count per category.

Endpoint

@router.get("/categories/{cat_id}/tree")
async def get_category_tree(cat_id: int, db = Depends(get_db)):
    """Subtree of a category with post count."""
    result = await db.execute(text("""
        WITH RECURSIVE subtree AS (
            SELECT id, name, parent_id, 0 AS depth
            FROM categories WHERE id = :cat_id
            UNION ALL
            SELECT c.id, c.name, c.parent_id, s.depth + 1
            FROM categories c JOIN subtree s ON c.parent_id = s.id
        )
        SELECT
            s.id, s.name, s.depth,
            COUNT(p.id) AS post_count
        FROM subtree s
        LEFT JOIN posts p ON p.category_id = s.id
        GROUP BY s.id, s.name, s.depth
        ORDER BY s.depth, s.name
    """), {"cat_id": cat_id})

    return result.mappings().all()

Breadcrumb endpoint

@router.get("/categories/{cat_id}/breadcrumb")
async def get_breadcrumb(cat_id: int, db = Depends(get_db)):
    """Path from root to this category."""
    result = await db.execute(text("""
        WITH RECURSIVE ancestors AS (
            SELECT id, name, parent_id, 0 AS depth
            FROM categories WHERE id = :cat_id
            UNION ALL
            SELECT c.id, c.name, c.parent_id, a.depth + 1
            FROM categories c JOIN ancestors a ON c.id = a.parent_id
        )
        SELECT name FROM ancestors ORDER BY depth DESC
    """), {"cat_id": cat_id})

    return [row[0] for row in result]

Filter posts by category including subcategories

@router.get("/posts/by-category/{cat_id}")
async def posts_by_category_recursive(cat_id: int, db = Depends(get_db)):
    """Posts in a category OR its subcategories."""
    result = await db.execute(text("""
        WITH RECURSIVE category_tree AS (
            SELECT id FROM categories WHERE id = :cat_id
            UNION ALL
            SELECT c.id FROM categories c
            JOIN category_tree ct ON c.parent_id = ct.id
        )
        SELECT p.id, p.title, p.published_at
        FROM posts p
        WHERE p.category_id IN (SELECT id FROM category_tree)
        ORDER BY p.published_at DESC
        LIMIT 50
    """), {"cat_id": cat_id})

    return result.mappings().all()

---

Component 6: advisory lock on the crons

Problem: two concurrent crons can run at the same time: MV refresh or FTS re-indexing. We need to guarantee a single instance.

Refresh cron with an advisory lock

# app/tasks/refresh_top_posts.py
import asyncio
from sqlalchemy import text
from app.database import SessionLocal
from app.services.advisory_locks import advisory_xact_lock_ns, LockNamespace


REFRESH_LOCK = hash("cron:refresh_top_posts") & 0x7FFFFFFF


async def refresh_top_posts():
    async with SessionLocal() as session:
        async with session.begin():
            async with advisory_xact_lock_ns(
                session, LockNamespace.REFRESH_MV, REFRESH_LOCK
            ) as got_lock:
                if not got_lock:
                    print("Refresh already running, skipping")
                    return

                await session.execute(text("""
                    REFRESH MATERIALIZED VIEW CONCURRENTLY top_posts_weekly
                """))
                print(f"Refreshed top_posts_weekly")


if __name__ == "__main__":
    asyncio.run(refresh_top_posts())

Crontab:

*/30 * * * * cd /app && python -m app.tasks.refresh_top_posts

FTS re-indexing cron with an advisory lock + savepoints

# app/tasks/reindex_fts.py
async def reindex_posts_fts():
    async with SessionLocal() as session:
        async with session.begin():
            async with advisory_xact_lock_ns(
                session, LockNamespace.CRON, hash("reindex_fts") & 0x7FFFFFFF
            ) as got_lock:
                if not got_lock:
                    print("Reindex already running")
                    return

                # The search vector is generated, it doesn't normally need reindex
                # But if you want to re-process (e.g. the dictionary changed), refresh the GIN
                await session.execute(text("REINDEX INDEX CONCURRENTLY idx_posts_search_vector"))


if __name__ == "__main__":
    asyncio.run(reindex_posts_fts())

REINDEX INDEX CONCURRENTLY is the equivalent operation for refreshing a GIN — it doesn't lock writes during the rebuild.

Bulk import of comments with savepoints

@router.post("/comments/bulk")
async def bulk_import_comments(items: list[dict], db = Depends(get_db)):
    success = 0
    errors = []

    for i, item in enumerate(items):
        try:
            async with db.begin_nested():
                comment = Comment(**item)
                db.add(comment)
            success += 1
        except IntegrityError as e:
            errors.append({"index": i, "error": str(e)})

    await db.commit()
    return {"success": success, "errors": errors}

Invalid items don't roll back the valid ones — a savepoint per item.

---

Final state of the project

6 components implemented:

1. ✅ JSONB metadata + GIN index (modules 1-2).
2. ✅ Spanish FTS + pg_trgm fallback (module 3).
3. ✅ Materialized view top_posts_weekly (module 5).
4. ✅ Partitioning of comments by month (module 4 + #13 module 5 zero-downtime).
5. ✅ Recursive categories with a CTE (module 8).
6. ✅ Advisory lock on the crons + savepoints in bulk (module 6).

Plus module 7 extensions:

• citext for users.email (case-insensitive).
• pg_trgm for the fuzzy fallback in search.

---

Final benchmarks (before/after)

Endpoint	Baseline	After Refactor	Improvement
Search	145ms (LIKE)	8ms (FTS)	18x
Popular posts	850ms (joins)	4ms (MV)	212x
By tag	N/A (new)	5ms (JSONB GIN)	new capability
Recent comments per post	18ms	4ms (partitioning)	4.5x
Category tree	N/A (new)	12ms (recursive CTE)	new capability
Breadcrumb	N/A (new)	6ms	new capability
Cron refresh	overlapped runs	safe (advisory lock)	reliability ↑
Bulk import	all-or-nothing	partial success	UX ↑

---

Integration tests

# tests/test_refactor.py

@pytest.mark.asyncio
async def test_jsonb_metadata_search(client):
    """A JSONB containment query uses the GIN index."""
    response = await client.get("/posts/by-tag/python")
    assert response.status_code == 200
    posts = response.json()
    assert all("python" in p["metadata"]["tags"] for p in posts)


@pytest.mark.asyncio
async def test_fts_search_finds_relevant(client):
    """FTS finds relevant results with stemming."""
    response = await client.get("/search?q=programando")
    assert response.status_code == 200
    body = response.json()
    # Should find posts with 'programa', 'programar', 'programación' (Spanish stemming)
    assert len(body["results"]) > 0


@pytest.mark.asyncio
async def test_search_with_typo_returns_suggestions(client):
    """Typo in the search → suggestions with pg_trgm."""
    response = await client.get("/search?q=Phyton")  # typo
    body = response.json()
    if len(body["results"]) < 5:
        assert len(body["suggestions"]) > 0
        assert any("python" in s["title"].lower() for s in body["suggestions"])


@pytest.mark.asyncio
async def test_top_posts_uses_mv(client):
    """The popular endpoint is fast (MV)."""
    import time
    start = time.perf_counter()
    response = await client.get("/posts/popular")
    elapsed = (time.perf_counter() - start) * 1000

    assert response.status_code == 200
    assert elapsed < 50  # MV should be <50ms


@pytest.mark.asyncio
async def test_category_tree_recursive(client):
    """The recursive CTE returns the complete subtree."""
    response = await client.get("/categories/1/tree")
    body = response.json()
    assert len(body) > 1  # has descendants
    depths = [c["depth"] for c in body]
    assert max(depths) > 0  # has levels


@pytest.mark.asyncio
async def test_advisory_lock_prevents_double_run(monkeypatch):
    """Two concurrent crons — only one runs."""
    from app.tasks.refresh_top_posts import refresh_top_posts

    results = await asyncio.gather(
        refresh_top_posts(),
        refresh_top_posts(),
        return_exceptions=True,
    )
    # Only one does the work; the other skips
    # (verify via logs or internal flags)


@pytest.mark.asyncio
async def test_bulk_partial_success(client):
    """Bulk with an invalid item — the valid ones succeed."""
    payload = [
        {"post_id": 1, "user_id": 1, "body": "Valid"},
        {"post_id": 99999999, "user_id": 1, "body": "Invalid FK"},
        {"post_id": 1, "user_id": 1, "body": "Valid 2"},
    ]
    response = await client.post("/comments/bulk", json=payload)
    body = response.json()
    assert body["success"] == 2
    assert len(body["errors"]) == 1

---

Common traps and mistakes in phase 2

1. Partitioning without a zero-downtime plan.

Migrating to a partitioned table in production requires the expand-contract pattern. Without it, downtime up to hours.

2. Recursive CTE without an index on parent_id.

Sequential scans on every iteration. The index is mandatory.

3. Session-level advisory lock + PgBouncer transaction mode.

As we saw in module 6, do NOT mix them. Use transaction-level with PgBouncer.

4. Cron without try/finally for the unlock.

If the cron crashes while holding the lock, it stays held. A context manager (async with advisory_xact_lock) handles it automatically.

5. Tests without a real Postgres.

RLS, triggers, partitioning, CTEs can't be mocked. A real Postgres (testcontainers) is mandatory.

6. Forgetting to refresh the MV.

The cron has to run. If the MV isn't refreshed, the data stays stale indefinitely. Verify the cron is active.

7. FTS re-indexing without REINDEX CONCURRENTLY.

REINDEX INDEX (without CONCURRENTLY) locks writes. For production, always CONCURRENTLY.

---

Summary and next step

What you have now:

• 6 components implemented.
• Zero-downtime migration executed (partitioning).
• Automated tests verifying each component.
• Before/after benchmarks documented.
• Crons protected with advisory locks.

Before moving on:

• Verify all tests pass.
• Verify the benchmarks are reproducible.
• Push to the repo.

In the next capsule we close the guide with the final documentation: BENCHMARKS.md with the complete table, ARCHITECTURE.md with ADRs per decision, updated README. And a reflection on the close of the Data Layer sub-track.

---

Resources

• Module 4 (partitioning) of this guide.
• Module 5 (zero-downtime migrations) of #13.
• Module 6 (advisory locks + savepoints) of this guide.
• Capsules 02-05 of this module (recursive CTEs).

---

Capsule 07 of 08 — Module 8 — Advanced PostgreSQL for Backend Guide