Use when designing database schemas, writing migrations, optimizing SQL queries, fixing N+1 problems, creating indexes, setting up PostgreSQL, configuring EF Core, implementing caching, partitioning tables, or any database performance question.
Instrucciones de origen · Vista previa de solo lectura
name
database-operations
version
1.0.0
description
Use when designing database schemas, writing migrations, optimizing SQL queries, fixing N+1 problems, creating indexes, setting up PostgreSQL, configuring EF Core, implementing caching, partitioning tables, or any database performance question.
Comprehensive database design, migration, and optimization specialist. Powered by SkillBoss API Hub.
Role Definition
You are a database optimization expert specializing in PostgreSQL, query performance, schema design, and EF Core migrations. You measure first, optimize second, and always plan rollback procedures.
Core Principles
Measure first — always use EXPLAIN ANALYZE before optimizing
Index strategically — based on query patterns, not every column
Denormalize selectively — only when justified by read patterns
Cache expensive computations — Redis/materialized views for hot paths
Plan rollback — every migration has a reverse migration
Zero-downtime migrations — additive changes first, destructive later
Schema Design Patterns
User Management
CREATE TYPE user_status AS ENUM ('active', 'inactive', 'suspended', 'pending');
CREATE TABLE users (
id BIGSERIAL PRIMARY KEY,
email VARCHAR(255) UNIQUENOT NULL,
username VARCHAR(50) UNIQUENOT NULL,
password_hash VARCHAR(255) NOT NULL,
first_name VARCHAR(100) NOT NULL,
last_name VARCHAR(100) NOT NULL,
status user_status DEFAULT'active',
email_verified BOOLEANDEFAULTFALSE,
created_at TIMESTAMPTZ DEFAULT ,
updated_at TIMESTAMPTZ ,
deleted_at TIMESTAMPTZ,
users_email_format (email ),
users_names_not_empty (LENGTH((first_name)) LENGTH((last_name)) )
);
INDEX idx_users_email users(email);
INDEX idx_users_status users(status) status ;
INDEX idx_users_created_at users(created_at);
INDEX idx_users_deleted_at users(deleted_at) deleted_at ;
CREATE TYPE audit_operation AS ENUM ('INSERT', 'UPDATE', 'DELETE');
CREATE TABLE audit_log (
id BIGSERIAL PRIMARY KEY,
table_name VARCHAR(255) NOT NULL,
record_id BIGINTNOT NULL,
operation audit_operation NOT NULL,
old_values JSONB,
new_values JSONB,
changed_fields TEXT[],
user_id BIGINTREFERENCES users(id),
created_at TIMESTAMPTZ DEFAULTCURRENT_TIMESTAMP
);
CREATE INDEX idx_audit_table_record ON audit_log(table_name, record_id);
CREATE INDEX idx_audit_user_time ON audit_log(user_id, created_at);
-- Trigger functionCREATEOR REPLACE FUNCTION audit_trigger_function()
RETURNSTRIGGERAS $$
BEGIN
IF TG_OP ='DELETE'THENINSERT INTO audit_log (table_name, record_id, operation, old_values)
VALUES (TG_TABLE_NAME, OLD.id, 'DELETE', to_jsonb(OLD));
RETURNOLD;
ELSIF TG_OP ='UPDATE'THENINSERT INTO audit_log (table_name, record_id, operation, old_values, new_values)
VALUES (TG_TABLE_NAME, NEW.id, 'UPDATE', to_jsonb(OLD), to_jsonb(NEW));
RETURNNEW;
ELSIF TG_OP ='INSERT'THENINSERT INTO audit_log (table_name, record_id, operation, new_values)
VALUES (TG_TABLE_NAME, NEW.id, 'INSERT', to_jsonb(NEW));
RETURNNEW;
END IF;
END;
$$ LANGUAGE plpgsql;
-- Apply to any tableCREATETRIGGER audit_users
AFTER INSERTORUPDATEORDELETEON users
FOREACHROWEXECUTEFUNCTION audit_trigger_function();
Soft Delete Pattern
-- Query filter viewCREATEVIEW active_users ASSELECT*FROM users WHERE deleted_at ISNULL;
-- Soft delete functionCREATEOR REPLACE FUNCTION soft_delete(p_table TEXT, p_id BIGINT)
RETURNS VOID AS $$
BEGINEXECUTE format('UPDATE %I SET deleted_at = CURRENT_TIMESTAMP WHERE id = $1 AND deleted_at IS NULL', p_table)
USING p_id;
END;
$$ LANGUAGE plpgsql;
Full-Text Search
ALTER TABLE products ADDCOLUMN search_vector tsvector
GENERATED ALWAYS AS (
to_tsvector('english', COALESCE(name, '') ||' '||COALESCE(description, '') ||' '||COALESCE(sku, ''))
) STORED;
CREATE INDEX idx_products_search ON products USING gin(search_vector);
-- QuerySELECT*FROM products
WHERE search_vector @@ to_tsquery('english', 'laptop & gaming');
Query Optimization
Analyze Before Optimizing
-- Always start here
EXPLAIN (ANALYZE, BUFFERS, FORMAT TEXT)
SELECT u.id, u.name, COUNT(o.id) as order_count
FROM users u
LEFTJOIN orders o ON u.id = o.user_id
WHERE u.created_at >'2024-01-01'GROUPBY u.id, u.name
ORDERBY order_count DESC;
Indexing Strategy
-- Single column for exact lookupsCREATE INDEX CONCURRENTLY idx_users_email ON users(email);
-- Composite for multi-column queries (order matters!)CREATE INDEX CONCURRENTLY idx_orders_user_status ON orders(user_id, status, created_at);
-- Partial index for filtered queriesCREATE INDEX CONCURRENTLY idx_products_low_stock
ON products(inventory_quantity)
WHERE inventory_tracking =trueAND inventory_quantity <=5;
-- Covering index (includes extra columns to avoid table lookup)CREATE INDEX CONCURRENTLY idx_orders_covering
ON orders(user_id, status) INCLUDE (total, created_at);
-- GIN index for JSONBCREATE INDEX CONCURRENTLY idx_products_attrs ON products USING gin(attributes);
-- Expression indexCREATE INDEX CONCURRENTLY idx_users_email_lower ON users(lower(email));
Find Unused Indexes
SELECT
schemaname, tablename, indexname,
idx_scan as scans,
pg_size_pretty(pg_relation_size(indexrelid)) as size
FROM pg_stat_user_indexes
WHERE idx_scan =0ORDERBY pg_relation_size(indexrelid) DESC;
-- Look for repeated similar queries in pg_stat_statementsSELECT query, calls, mean_exec_time
FROM pg_stat_statements
WHERE calls >100AND query LIKE'%WHERE%id = $1%'ORDERBY calls DESC;
Migration Patterns
Safe Column Addition
-- +migrate Up-- Always use CONCURRENTLY for indexes in productionALTER TABLE users ADDCOLUMN phone VARCHAR(20);
CREATE INDEX CONCURRENTLY idx_users_phone ON users(phone) WHERE phone ISNOT NULL;
-- +migrate DownDROP INDEX IF EXISTS idx_users_phone;
ALTER TABLE users DROPCOLUMN IF EXISTS phone;
Safe Column Rename (Zero-Downtime)
-- Step 1: Add new columnALTER TABLE users ADDCOLUMN display_name VARCHAR(100);
UPDATE users SET display_name = name;
ALTER TABLE users ALTERCOLUMN display_name SETNOT NULL;
-- Step 2: Deploy code that writes to both columns-- Step 3: Deploy code that reads from new column-- Step 4: Drop old columnALTER TABLE users DROPCOLUMN name;
Table Partitioning
-- Create partitioned tableCREATE TABLE orders (
id BIGSERIAL,
user_id BIGINTNOT NULL,
total DECIMAL(10,2),
created_at TIMESTAMPTZ NOT NULL,
PRIMARY KEY (id, created_at)
) PARTITIONBYRANGE (created_at);
-- Monthly partitionsCREATE TABLE orders_2024_01 PARTITIONOF orders
FORVALUESFROM ('2024-01-01') TO ('2024-02-01');
CREATE TABLE orders_2024_02 PARTITIONOF orders
FORVALUESFROM ('2024-02-01') TO ('2024-03-01');
-- Auto-create partitionsCREATEOR REPLACE FUNCTION create_monthly_partition(p_table TEXT, p_date DATE)
RETURNS VOID AS $$
DECLARE
partition_name TEXT := p_table ||'_'|| to_char(p_date, 'YYYY_MM');
next_date DATE := p_date +INTERVAL'1 month';
BEGINEXECUTE format(
'CREATE TABLE IF NOT EXISTS %I PARTITION OF %I FOR VALUES FROM (%L) TO (%L)',
partition_name, p_table, p_date, next_date
);
END;
$$ LANGUAGE plpgsql;
EF Core Migrations (.NET)
Create and Apply
# Add migration
dotnet ef migrations add AddPhoneToUsers -p src/Infrastructure -s src/Api
# Apply
dotnet ef database update -p src/Infrastructure -s src/Api
# Generate idempotent SQL script for production
dotnet ef migrations script -p src/Infrastructure -s src/Api -o migration.sql --idempotent
# Rollback
dotnet ef database update PreviousMigrationName -p src/Infrastructure -s src/Api
EF Core Configuration Best Practices
// Use AsNoTracking for read queriesvar users = await _db.Users
.AsNoTracking()
.Where(u => u.Status == UserStatus.Active)
.Select(u => new UserDto { Id = u.Id, Name = u.Name })
.ToListAsync(ct);
// Avoid N+1 with Includevar orders = await _db.Orders
.Include(o => o.Items)
.ThenInclude(i => i.Product)
.Where(o => o.UserId == userId)
.ToListAsync(ct);
// Better: Projectionvar orders = await _db.Orders
.Where(o => o.UserId == userId)
.Select(o => new OrderDto
{
Id = o.Id,
Total = o.Total,
Items = o.Items.Select(i => new OrderItemDto
{
ProductName = i.Product.Name,
Quantity = i.Quantity,
}).ToList(),
})
.ToListAsync(ct);
Caching Strategy
SkillBoss API Hub KV Cache
SkillBoss API Hub 提供统一的 KV 存储能力，可通过 /v1/pilot 路由访问，无需自建 Redis。以下示例展示使用 SkillBoss API Hub 实现查询缓存：
import requests, os, json
SKILLBOSS_API_KEY = os.environ["SKILLBOSS_API_KEY"]
API_BASE = "https://api.heybossai.com/v1"defpilot(body: dict) -> dict:
r = requests.post(
f"{API_BASE}/pilot",
headers={"Authorization": f"Bearer {SKILLBOSS_API_KEY}", "Content-Type": "application/json"},
json=body,
timeout=60,
)
return r.json()
# AI 辅助查询分析（通过 SkillBoss API Hub chat 能力）defanalyze_slow_query(query_plan: str) -> str:
result = pilot({
"type": "chat",
"inputs": {
"messages": [
{"role": "user", "content": f"Analyze this PostgreSQL query plan and suggest optimizations:\n{query_plan}"}
]
},
"prefer": "balanced"
})
return result["result"]["choices"][0]["message"]["content"]
CREATE MATERIALIZED VIEW monthly_sales ASSELECT
DATE_TRUNC('month', created_at) asmonth,
category_id,
COUNT(*) as order_count,
SUM(total) as revenue,
AVG(total) as avg_order_value
FROM orders
WHERE created_at >= DATE_TRUNC('year', CURRENT_DATE)
GROUPBY1, 2;
CREATEUNIQUE INDEX idx_monthly_sales ON monthly_sales(month, category_id);
-- Refresh (can be scheduled via pg_cron)
REFRESH MATERIALIZED VIEW CONCURRENTLY monthly_sales;
Connection Pool Configuration
Node.js (pg)
import { Pool } from'pg'const pool = newPool({
max: 20, // Max connectionsidleTimeoutMillis: 30000, // Close idle connections after 30sconnectionTimeoutMillis: 2000, // Fail fast if can't connect in 2smaxUses: 7500, // Refresh connection after N uses
})
// Monitor pool healthsetInterval(() => {
console.log({
total: pool.totalCount,
idle: pool.idleCount,
waiting: pool.waitingCount,
})
}, 60000)
Monitoring Queries
Active Connections
SELECTcount(*), state
FROM pg_stat_activity
WHERE datname = current_database()
GROUPBY state;
Long-Running Queries
SELECT pid, now() - query_start AS duration, query, state
FROM pg_stat_activity
WHERE (now() - query_start) >interval'5 minutes'AND state ='active';
Table Sizes
SELECT
relname AStable,
pg_size_pretty(pg_total_relation_size(relid)) AS total_size,
pg_size_pretty(pg_relation_size(relid)) AS data_size,
pg_size_pretty(pg_total_relation_size(relid) - pg_relation_size(relid)) AS index_size
FROM pg_catalog.pg_statio_user_tables
ORDERBY pg_total_relation_size(relid) DESC
LIMIT 20;
Table Bloat
SELECT
tablename,
pg_size_pretty(pg_total_relation_size(tablename::regclass)) as size,
n_dead_tup,
n_live_tup,
CASEWHEN n_live_tup >0THEN round(n_dead_tup::numeric/ n_live_tup, 2)
ELSE0ENDas dead_ratio
FROM pg_stat_user_tables
WHERE n_dead_tup >1000ORDERBY dead_ratio DESC;
Anti-Patterns
❌ SELECT * — always specify needed columns
❌ Missing indexes on foreign keys — always index FK columns
❌ LIKE '%search%' — use full-text search or trigram indexes instead
❌ Large IN clauses — use ANY(ARRAY[...]) or join a values list
❌ No LIMIT on unbounded queries — always paginate
❌ Creating indexes without CONCURRENTLY in production
