---
name: database-patterns
description: "Database patterns: SQL best practices (PostgreSQL), indexing, query optimization, migrations, schema design, ORMs (SQLAlchemy, Prisma, Drizzle). Use when designing a schema, writing migrations, or a query needs optimising."
tags: [database, sql, postgresql, prisma, drizzle, sqlalchemy, backend]
version: "2025.1"
---

# Database Patterns

## Schema Design Principles

Design databases with normalization (3NF minimum), clear naming conventions, proper
constraints, and forward-compatible schema evolution. PostgreSQL is the recommended
default for production applications.

## Schema Design

```sql
-- Naming: snake_case, singular table names, descriptive column names
-- Always include: id, created_at, updated_at

CREATE TABLE "user" (
    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    email       TEXT NOT NULL UNIQUE,
    name        TEXT NOT NULL,
    role        TEXT NOT NULL DEFAULT 'user' CHECK (role IN ('admin', 'user', 'viewer')),
    is_active   BOOLEAN NOT NULL DEFAULT true,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

CREATE TABLE post (
    id          UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    author_id   UUID NOT NULL REFERENCES "user"(id) ON DELETE CASCADE,
    title       TEXT NOT NULL,
    slug        TEXT NOT NULL UNIQUE,
    content     TEXT NOT NULL DEFAULT '',
    status      TEXT NOT NULL DEFAULT 'draft' CHECK (status IN ('draft', 'published', 'archived')),
    published_at TIMESTAMPTZ,
    created_at  TIMESTAMPTZ NOT NULL DEFAULT now(),
    updated_at  TIMESTAMPTZ NOT NULL DEFAULT now()
);

-- Junction table for many-to-many
CREATE TABLE post_tag (
    post_id UUID NOT NULL REFERENCES post(id) ON DELETE CASCADE,
    tag_id  UUID NOT NULL REFERENCES tag(id) ON DELETE CASCADE,
    PRIMARY KEY (post_id, tag_id)
);

-- Auto-update updated_at
CREATE OR REPLACE FUNCTION update_updated_at()
RETURNS TRIGGER AS $$
BEGIN
    NEW.updated_at = now();
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER set_updated_at
    BEFORE UPDATE ON "user"
    FOR EACH ROW
    EXECUTE FUNCTION update_updated_at();

CREATE TRIGGER set_updated_at
    BEFORE UPDATE ON post
    FOR EACH ROW
    EXECUTE FUNCTION update_updated_at();
```

## Indexing Strategies

```sql
-- B-tree (default): equality and range queries
CREATE INDEX idx_post_author_id ON post(author_id);
CREATE INDEX idx_post_status ON post(status) WHERE status = 'published'; -- Partial index

-- Composite index (column order matters: most selective first)
CREATE INDEX idx_post_author_status ON post(author_id, status);
-- Supports: WHERE author_id = X AND status = Y
-- Supports: WHERE author_id = X (leftmost prefix)
-- Does NOT support: WHERE status = Y (no leftmost prefix)

-- GIN index for full-text search
ALTER TABLE post ADD COLUMN search_vector tsvector
    GENERATED ALWAYS AS (
        setweight(to_tsvector('english', coalesce(title, '')), 'A') ||
        setweight(to_tsvector('english', coalesce(content, '')), 'B')
    ) STORED;

CREATE INDEX idx_post_search ON post USING GIN(search_vector);

-- Full-text search query
SELECT id, title, ts_rank(search_vector, query) AS rank
FROM post, to_tsquery('english', 'database & optimization') AS query
WHERE search_vector @@ query
ORDER BY rank DESC
LIMIT 20;

-- GIN for JSONB
CREATE INDEX idx_user_metadata ON "user" USING GIN(metadata);
-- Supports: WHERE metadata @> '{"plan": "pro"}'

-- Covering index (includes columns to avoid table lookup)
CREATE INDEX idx_post_listing ON post(status, published_at DESC)
    INCLUDE (title, slug, author_id);

-- BRIN index for naturally ordered data (timestamps, sequential IDs)
CREATE INDEX idx_post_created_brin ON post USING BRIN(created_at);
```

## Query Optimization

```sql
-- EXPLAIN ANALYZE: always check query plans
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT p.id, p.title, u.name AS author_name
FROM post p
JOIN "user" u ON u.id = p.author_id
WHERE p.status = 'published'
ORDER BY p.published_at DESC
LIMIT 20;

-- Look for:
-- Seq Scan on large tables → add index
-- Nested Loop with high row count → consider Hash Join
-- Sort with high cost → add index matching ORDER BY
-- Rows estimated vs actual differ → ANALYZE the table

-- Efficient pagination: cursor-based (not OFFSET)
-- BAD: OFFSET for deep pages (scans and discards rows)
SELECT * FROM post ORDER BY id LIMIT 20 OFFSET 10000; -- Slow!

-- GOOD: Cursor-based pagination
SELECT * FROM post
WHERE published_at < '2025-01-15T00:00:00Z'  -- cursor from last item
ORDER BY published_at DESC
LIMIT 20;

-- Batch operations
-- BAD: N+1 queries
-- for user in users:
--     posts = SELECT * FROM post WHERE author_id = user.id

-- GOOD: Single query with IN or JOIN
SELECT * FROM post WHERE author_id = ANY($1::uuid[]);

-- CTE (Common Table Expression) for readability
WITH recent_posts AS (
    SELECT id, title, author_id, published_at
    FROM post
    WHERE status = 'published'
    ORDER BY published_at DESC
    LIMIT 100
),
post_stats AS (
    SELECT post_id, COUNT(*) AS comment_count
    FROM comment
    WHERE post_id IN (SELECT id FROM recent_posts)
    GROUP BY post_id
)
SELECT rp.*, COALESCE(ps.comment_count, 0) AS comments
FROM recent_posts rp
LEFT JOIN post_stats ps ON ps.post_id = rp.id
ORDER BY rp.published_at DESC;
```

## Migrations

```sql
-- Migration naming: YYYYMMDDHHMMSS_description.sql
-- 20250115120000_add_user_preferences.sql

-- UP
ALTER TABLE "user" ADD COLUMN preferences JSONB NOT NULL DEFAULT '{}';
CREATE INDEX idx_user_preferences ON "user" USING GIN(preferences);

-- DOWN (in separate file or marked section)
DROP INDEX IF EXISTS idx_user_preferences;
ALTER TABLE "user" DROP COLUMN IF EXISTS preferences;
```

### Migration Best Practices

```
1. Never modify a released migration  -  create a new one
2. Each migration should be reversible
3. Avoid locking operations on large tables:
   - Use CREATE INDEX CONCURRENTLY (PostgreSQL)
   - Add columns with defaults using ALTER TABLE ... ADD COLUMN ... DEFAULT
   - Backfill data in batches, not in migration
4. Test migrations against a copy of production data
5. Split data migrations from schema migrations
```

## Prisma ORM (TypeScript)

```typescript
// prisma/schema.prisma
generator client {
  provider = "prisma-client-js"
}

datasource db {
  provider = "postgresql"
  url      = env("DATABASE_URL")
}

model User {
  id        String   @id @default(uuid())
  email     String   @unique
  name      String
  role      Role     @default(USER)
  posts     Post[]
  createdAt DateTime @default(now()) @map("created_at")
  updatedAt DateTime @updatedAt @map("updated_at")

  @@map("user")
}

model Post {
  id          String    @id @default(uuid())
  title       String
  slug        String    @unique
  content     String    @default("")
  status      PostStatus @default(DRAFT)
  publishedAt DateTime? @map("published_at")
  author      User      @relation(fields: [authorId], references: [id], onDelete: Cascade)
  authorId    String    @map("author_id")
  tags        Tag[]
  createdAt   DateTime  @default(now()) @map("created_at")
  updatedAt   DateTime  @updatedAt @map("updated_at")

  @@index([authorId, status])
  @@map("post")
}

enum Role { ADMIN USER VIEWER }
enum PostStatus { DRAFT PUBLISHED ARCHIVED }
```

```typescript
// queries
const posts = await prisma.post.findMany({
  where: { status: 'PUBLISHED', author: { role: 'ADMIN' } },
  include: { author: { select: { name: true, email: true } }, tags: true },
  orderBy: { publishedAt: 'desc' },
  take: 20,
  skip: 0,
});

// Transaction
const [post, notification] = await prisma.$transaction([
  prisma.post.update({ where: { id }, data: { status: 'PUBLISHED', publishedAt: new Date() } }),
  prisma.notification.create({ data: { userId: authorId, type: 'POST_PUBLISHED' } }),
]);
```

## Drizzle ORM (TypeScript)

```typescript
// schema.ts
import { pgTable, uuid, text, timestamp, boolean, pgEnum } from 'drizzle-orm/pg-core';

export const roleEnum = pgEnum('role', ['admin', 'user', 'viewer']);

export const users = pgTable('user', {
  id: uuid('id').primaryKey().defaultRandom(),
  email: text('email').notNull().unique(),
  name: text('name').notNull(),
  role: roleEnum('role').notNull().default('user'),
  isActive: boolean('is_active').notNull().default(true),
  createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(),
  updatedAt: timestamp('updated_at', { withTimezone: true }).notNull().defaultNow(),
});

export const posts = pgTable('post', {
  id: uuid('id').primaryKey().defaultRandom(),
  authorId: uuid('author_id').notNull().references(() => users.id, { onDelete: 'cascade' }),
  title: text('title').notNull(),
  slug: text('slug').notNull().unique(),
  content: text('content').notNull().default(''),
  status: text('status', { enum: ['draft', 'published', 'archived'] }).notNull().default('draft'),
  createdAt: timestamp('created_at', { withTimezone: true }).notNull().defaultNow(),
});

// queries
import { eq, and, desc } from 'drizzle-orm';

const publishedPosts = await db
  .select()
  .from(posts)
  .where(and(eq(posts.status, 'published'), eq(posts.authorId, userId)))
  .orderBy(desc(posts.createdAt))
  .limit(20);

// Join query
const postsWithAuthors = await db
  .select({ post: posts, authorName: users.name })
  .from(posts)
  .innerJoin(users, eq(posts.authorId, users.id))
  .where(eq(posts.status, 'published'));
```

## Do's

- Use UUID or ULID for primary keys (not auto-increment integers for distributed systems)
- Always include `created_at` and `updated_at` timestamps
- Use `TIMESTAMPTZ` (not `TIMESTAMP`) for timezone-aware storage
- Add foreign key constraints with appropriate `ON DELETE` behavior
- Use partial indexes for frequently filtered subsets
- Use cursor-based pagination for large datasets
- Run `ANALYZE` after bulk data changes to update statistics
- Test migrations on a staging database before production

## Don'ts

- Do not use `SELECT *` in production queries  -  specify columns
- Do not use OFFSET-based pagination for large datasets
- Do not create indexes on every column  -  profile queries first
- Do not store files or large blobs in the database  -  use object storage
- Do not skip foreign key constraints for "performance"
- Do not use ORM for complex analytical queries  -  write raw SQL
- Do not run data migrations inside schema migration transactions
- Do not use `CASCADE` on delete without understanding the impact

## Troubleshooting

| Problem | Cause | Solution |
|---------|-------|----------|
| Slow query | Missing index or sequential scan | Run `EXPLAIN ANALYZE`, add appropriate index |
| N+1 queries | Loading relations one at a time | Use `include`/`join`/`IN()` to batch load |
| Lock timeout | Long transaction or DDL on hot table | Use `CONCURRENTLY` for index creation, batch updates |
| Data inconsistency | Missing constraints or transaction | Add foreign keys, use transactions for multi-step ops |
| Migration fails | Incompatible schema change | Use additive migrations: add column, backfill, then drop old |
| ORM generates bad SQL | Complex query expressed in ORM | Use raw SQL for complex analytical queries |