
Database Designer
- 140 installs
- 451 repo stars
- Updated July 21, 2026
- borghei/claude-skills
Design normalized schemas, indexes, constraints, and migration plans for new features needing durable storage, reporting, and transactional integrity.
About
Helps Claude architect production-grade relational databases with clear entities, keys, constraints, indexing, and migration steps aligned to SaaS and API backends needing scalable persistence.
- Entity-relationship modeling
- Normalization and constraint design
- Index and query-plan awareness
- Migration and rollout planning
- Multi-tenant and audit patterns
Database Designer by the numbers
- 140 all-time installs (skills.sh)
- Ranked #284 of 911 Databases skills by installs in the Skillselion catalog
- Data as of Aug 5, 2026 (Skillselion catalog sync)
npx skills add https://github.com/borghei/claude-skills --skill database-designerAdd your badge
Show developers this skill is listed on Skillselion. Paste this into your README.
| Installs | 140 |
|---|---|
| repo stars | ★ 451 |
| Last updated | July 21, 2026 |
| Repository | borghei/claude-skills ↗ |
What it does
Design normalized schemas, indexes, constraints, and migration plans for new features needing durable storage, reporting, and transactional integrity.
Files
Database Designer
The agent analyzes SQL schemas for normalization compliance, recommends optimal indexes based on query patterns, and generates safe migration scripts with rollback procedures. It produces Mermaid ERDs, detects redundant indexes, and implements zero-downtime expand-contract migration patterns for PostgreSQL and MySQL.
Quick Start
# Analyze a schema for normalization issues and generate ERD
python schema_analyzer.py --input schema.sql --generate-erd --output-format json
# Recommend indexes based on query patterns
python index_optimizer.py --schema schema.json --queries queries.json --analyze-existing
# Generate migration scripts between schema versions
python migration_generator.py --current current.json --target target.json --zero-downtime---
Core Workflows
Workflow 1: Analyze and Optimize a Schema
1. Provide DDL (SQL) or JSON schema definition 2. Run schema_analyzer.py to detect normalization violations (1NF-BCNF), missing constraints, and naming issues 3. Review generated Mermaid ERD for relationship visualization 4. Run index_optimizer.py with query patterns to get index recommendations 5. Validation checkpoint: All 1NF-3NF violations addressed; foreign keys declared; no redundant indexes
python schema_analyzer.py -i schema.sql -f json -e -o report.json
python index_optimizer.py -s schema.json -q queries.json -e -p 2 -o index_report.jsonWorkflow 2: Generate a Safe Migration
1. Export current and target schemas as JSON 2. Run migration_generator.py to produce forward and rollback SQL 3. For large tables (10M+ rows), add --zero-downtime for expand-contract pattern 4. Review validation queries that confirm migration success 5. Validation checkpoint: Every forward step has a rollback counterpart; validation queries pass on test data
python migration_generator.py -c current.json -t target.json -z --include-validations -f json -o plan.jsonWorkflow 3: Index Optimization for Query Patterns
1. Document top 10 query patterns as JSON (WHERE clauses, JOINs, ORDER BY) 2. Run index_optimizer.py with --analyze-existing to find redundancies 3. Review composite index column ordering (most selective first) 4. Check for covering index opportunities 5. Validation checkpoint: Query patterns covered; no overlapping indexes; estimated 40%+ query time reduction
---
Index Type Selection
| Index Type | Best For | Example |
|---|---|---|
| B-tree | Range queries, sorting, equality | CREATE INDEX idx ON tasks (status, created_date) |
| Partial | Subset queries on hot data | CREATE INDEX idx ON users (email) WHERE status = 'active' |
| Covering | Avoiding table lookups | CREATE INDEX idx ON users (email) INCLUDE (name, status) |
| Hash | Exact match only | Primary keys, cache keys |
| GIN | JSONB, array, full-text | CREATE INDEX idx ON docs USING GIN (data) |
---
Anti-Patterns
- Over-indexing -- every column indexed wastes write performance and storage; index only columns appearing in WHERE, JOIN, and ORDER BY
- Missing foreign keys -- relying on application-layer referential integrity leads to orphaned records; always declare FK constraints
- VARCHAR(255) everywhere -- oversized columns waste memory in indexes; right-size columns based on actual data
- Premature denormalization -- denormalize only when EXPLAIN ANALYZE shows join-related bottlenecks, not preemptively
- Direct ALTER on large tables --
ALTER TABLE ... SET NOT NULLon a 100M-row table locks the table; use expand-contract pattern - No validation queries in migrations -- migrations without post-step validation risk silent data corruption
Troubleshooting
| Problem | Cause | Solution |
|---|---|---|
| Schema analyzer reports false 1NF violations | JSON or array columns detected as multi-valued fields | Review flagged columns; intentional JSONB/array usage is valid for document-style storage patterns |
| Index optimizer recommends indexes on low-selectivity columns | Boolean or status columns appear in frequent WHERE clauses | Use partial indexes (WHERE status = 'active') instead of full-column indexes to reduce overhead |
| Migration generator produces high-risk steps for column type changes | Direct ALTER COLUMN ... TYPE can lock tables and fail on incompatible data | Use the --zero-downtime flag to generate expand-contract migration patterns with safe backfill steps |
| ERD output missing relationships | Foreign key constraints not declared in DDL or JSON input | Ensure all FK relationships are explicitly defined; the analyzer only detects declared constraints |
| Composite index column order seems wrong | Optimizer orders by estimated selectivity, not query clause order | Verify cardinality estimates in the schema JSON; provide cardinality_estimate per column for accurate ordering |
| Redundancy analysis flags covering indexes as overlapping | Overlap ratio calculation uses Jaccard similarity on column sets | Review flagged pairs manually; covering indexes with INCLUDE columns serve a different purpose than their subsets |
| Validation queries fail after migration | Target schema JSON does not match actual post-migration state | Run --validate-only before and after migration; ensure the target JSON reflects all intended changes precisely |
Success Criteria
- Schema analysis detects 90%+ of normalization violations (1NF through BCNF) when provided complete DDL input
- Index recommendations reduce query execution time by 40%+ for analyzed query patterns (measured via EXPLAIN ANALYZE before/after)
- Migration scripts execute with zero data loss and include verified rollback for every forward step
- ERD generation produces valid Mermaid diagrams that render correctly for schemas with up to 50 tables
- Redundant index detection identifies 95%+ of duplicate and overlapping indexes with less than 5% false positive rate
- Zero-downtime migrations maintain full application availability during schema changes on tables with 10M+ rows
- Generated SQL statements are syntactically valid and compatible with PostgreSQL 14+ and MySQL 8.0+
Scope & Limitations
Covers:
- Schema design analysis for SQL databases (PostgreSQL, MySQL) including normalization, constraints, naming, and data types
- Index optimization with selectivity estimation, composite index ordering, covering indexes, and redundancy detection
- Migration generation with forward/rollback scripts, zero-downtime patterns, and validation queries
- ERD generation in Mermaid format from DDL or JSON schema definitions
Does NOT cover:
- Runtime query performance monitoring or live database profiling (see
performance-profilerskill) - NoSQL-specific schema design for MongoDB, DynamoDB, or Cassandra (conceptual guidance only in the reference sections)
- Database administration tasks such as backup/restore, replication setup, or user/role management
- Application-level ORM configuration, connection pool tuning, or driver-specific optimizations (see
database-schema-designerfor ORM-adjacent patterns)
Integration Points
| Skill | Integration | Data Flow |
|---|---|---|
migration-architect | Migration strategy and execution planning for large-scale schema changes | Database Designer generates migration SQL; Migration Architect orchestrates multi-service deployment order and rollback coordination |
database-schema-designer | Complementary schema design with focus on application-layer patterns | Database Designer provides normalization analysis; Schema Designer applies ORM mapping and application modeling conventions |
performance-profiler | Runtime validation of index and schema optimization recommendations | Database Designer outputs recommended indexes; Performance Profiler measures actual query plan improvements via EXPLAIN ANALYZE |
api-design-reviewer | Alignment between database schema and API resource contracts | Database Designer defines table structures; API Design Reviewer validates that endpoint schemas match underlying data models |
ci-cd-pipeline-builder | Automated migration execution in deployment pipelines | Database Designer generates migration scripts; CI/CD Pipeline Builder integrates them into deployment stages with validation gates |
observability-designer | Database performance monitoring and alerting post-optimization | Database Designer identifies query patterns; Observability Designer configures slow query alerts and index usage dashboards |
Tool Reference
schema_analyzer.py
Purpose: Analyzes SQL DDL statements and JSON schema definitions for normalization compliance, missing constraints, data type issues, naming convention violations, and relationship mapping. Generates Mermaid ERD diagrams.
Usage:
python schema_analyzer.py --input schema.sql --output-format json
python schema_analyzer.py --input schema.json --output-format text
python schema_analyzer.py --input schema.sql --generate-erd --output analysis.json
python schema_analyzer.py --input schema.sql --erd-onlyFlags/Parameters:
| Flag | Short | Required | Description |
|---|---|---|---|
--input | -i | Yes | Input file path (SQL DDL or JSON schema) |
--output | -o | No | Output file path (default: stdout) |
--output-format | -f | No | Output format: json or text (default: text) |
--generate-erd | -e | No | Include Mermaid ERD diagram in output |
--erd-only | No | Output only the Mermaid ERD diagram |
Example:
python schema_analyzer.py -i my_schema.sql -f json -e -o report.jsonOutput Formats:
text-- Human-readable report with normalization findings, constraint issues, data type recommendations, and naming violationsjson-- Structured JSON withnormalization_issues,constraint_issues,data_type_issues,naming_issues,relationships, and optionalerd_diagramfields
---
index_optimizer.py
Purpose: Analyzes schema definitions and query patterns to recommend optimal indexes. Identifies missing indexes, detects redundant and overlapping indexes, suggests composite index column ordering, estimates selectivity, and generates CREATE INDEX statements.
Usage:
python index_optimizer.py --schema schema.json --queries queries.json --format text
python index_optimizer.py --schema schema.json --queries queries.json --output recommendations.json --format json
python index_optimizer.py --schema schema.json --queries queries.json --analyze-existing
python index_optimizer.py --schema schema.json --queries queries.json --min-priority 2Flags/Parameters:
| Flag | Short | Required | Description |
|---|---|---|---|
--schema | -s | Yes | Schema definition JSON file |
--queries | -q | Yes | Query patterns JSON file |
--output | -o | No | Output file path (default: stdout) |
--format | -f | No | Output format: json or text (default: text) |
--analyze-existing | -e | No | Include analysis of existing indexes for redundancy |
--min-priority | -p | No | Minimum priority level to include: 1=highest, 4=lowest (default: 4) |
Example:
python index_optimizer.py -s schema.json -q queries.json -f json -e -p 2 -o index_report.jsonOutput Formats:
text-- Human-readable report with analysis summary, high-priority recommendations, redundancy issues, performance impact analysis, and CREATE INDEX statementsjson-- Structured JSON withanalysis_summary,index_recommendations(by priority),redundancy_analysis,size_estimates,sql_statements, andperformance_impactfields
---
migration_generator.py
Purpose: Generates safe migration scripts between schema versions. Compares current and target schemas, produces ALTER TABLE statements, implements zero-downtime expand-contract patterns, creates rollback scripts, and generates validation queries.
Usage:
python migration_generator.py --current current.json --target target.json --format text
python migration_generator.py --current current.json --target target.json --output migration.sql --format sql
python migration_generator.py --current current.json --target target.json --zero-downtime --format json
python migration_generator.py --current current.json --target target.json --validate-onlyFlags/Parameters:
| Flag | Short | Required | Description |
|---|---|---|---|
--current | -c | Yes | Current schema JSON file |
--target | -t | Yes | Target schema JSON file |
--output | -o | No | Output file path (default: stdout) |
--format | -f | No | Output format: json, text, or sql (default: text) |
--zero-downtime | -z | No | Generate zero-downtime migration using expand-contract pattern |
--validate-only | -v | No | Only generate validation queries, skip migration steps |
--include-validations | No | Include validation queries in migration output |
Example:
python migration_generator.py -c current.json -t target.json -z --include-validations -f json -o migration_plan.jsonOutput Formats:
text-- Human-readable migration plan with ordered steps, forward SQL, rollback SQL, risk levels, and execution timelinejson-- Structured JSON withmigration_id,steps(each withsql_forward,sql_rollback,validation_sql,risk_level,zero_downtime_phase),summary,execution_order, androllback_ordersql-- Raw SQL output with forward migration statements, suitable for direct execution or piping into a database client
{
"queries": [
{
"id": "user_login",
"type": "SELECT",
"table": "users",
"description": "User authentication lookup by email",
"where_conditions": [
{
"column": "email",
"operator": "=",
"selectivity": 0.95
}
],
"join_conditions": [],
"order_by": [],
"group_by": [],
"frequency": 5000,
"avg_execution_time_ms": 2.5
},
{
"id": "product_search_category",
"type": "SELECT",
"table": "products",
"description": "Product search within category with pagination",
"where_conditions": [
{
"column": "category_id",
"operator": "=",
"selectivity": 0.2
},
{
"column": "is_active",
"operator": "=",
"selectivity": 0.1
}
],
"join_conditions": [],
"order_by": [
{"column": "created_at", "direction": "DESC"}
],
"group_by": [],
"frequency": 2500,
"avg_execution_time_ms": 15.2
},
{
"id": "product_search_price_range",
"type": "SELECT",
"table": "products",
"description": "Product search by price range and brand",
"where_conditions": [
{
"column": "price",
"operator": "BETWEEN",
"selectivity": 0.3
},
{
"column": "brand",
"operator": "=",
"selectivity": 0.05
},
{
"column": "is_active",
"operator": "=",
"selectivity": 0.1
}
],
"join_conditions": [],
"order_by": [
{"column": "price", "direction": "ASC"}
],
"group_by": [],
"frequency": 800,
"avg_execution_time_ms": 25.7
},
{
"id": "user_orders_history",
"type": "SELECT",
"table": "orders",
"description": "User order history with pagination",
"where_conditions": [
{
"column": "user_id",
"operator": "=",
"selectivity": 0.8
}
],
"join_conditions": [],
"order_by": [
{"column": "created_at", "direction": "DESC"}
],
"group_by": [],
"frequency": 1200,
"avg_execution_time_ms": 8.3
},
{
"id": "order_details_with_items",
"type": "SELECT",
"table": "orders",
"description": "Order details with order items (JOIN query)",
"where_conditions": [
{
"column": "id",
"operator": "=",
"selectivity": 1.0
}
],
"join_conditions": [
{
"local_column": "id",
"foreign_table": "order_items",
"foreign_column": "order_id",
"join_type": "INNER"
}
],
"order_by": [],
"group_by": [],
"frequency": 3000,
"avg_execution_time_ms": 12.1
},
{
"id": "pending_orders_processing",
"type": "SELECT",
"table": "orders",
"description": "Processing queue - pending orders by date",
"where_conditions": [
{
"column": "status",
"operator": "=",
"selectivity": 0.15
},
{
"column": "created_at",
"operator": ">=",
"selectivity": 0.3
}
],
"join_conditions": [],
"order_by": [
{"column": "created_at", "direction": "ASC"}
],
"group_by": [],
"frequency": 150,
"avg_execution_time_ms": 45.2
},
{
"id": "user_orders_by_status",
"type": "SELECT",
"table": "orders",
"description": "User orders filtered by status",
"where_conditions": [
{
"column": "user_id",
"operator": "=",
"selectivity": 0.8
},
{
"column": "status",
"operator": "IN",
"selectivity": 0.4
}
],
"join_conditions": [],
"order_by": [
{"column": "created_at", "direction": "DESC"}
],
"group_by": [],
"frequency": 600,
"avg_execution_time_ms": 18.5
},
{
"id": "product_reviews_summary",
"type": "SELECT",
"table": "product_reviews",
"description": "Product review aggregation",
"where_conditions": [
{
"column": "product_id",
"operator": "=",
"selectivity": 0.85
}
],
"join_conditions": [],
"order_by": [],
"group_by": ["product_id"],
"frequency": 1800,
"avg_execution_time_ms": 22.3
},
{
"id": "inventory_low_stock",
"type": "SELECT",
"table": "products",
"description": "Low inventory alert query",
"where_conditions": [
{
"column": "inventory_count",
"operator": "<=",
"selectivity": 0.1
},
{
"column": "is_active",
"operator": "=",
"selectivity": 0.1
}
],
"join_conditions": [],
"order_by": [
{"column": "inventory_count", "direction": "ASC"}
],
"group_by": [],
"frequency": 50,
"avg_execution_time_ms": 35.8
},
{
"id": "popular_products_by_category",
"type": "SELECT",
"table": "order_items",
"description": "Popular products analysis with category join",
"where_conditions": [
{
"column": "created_at",
"operator": ">=",
"selectivity": 0.2
}
],
"join_conditions": [
{
"local_column": "product_id",
"foreign_table": "products",
"foreign_column": "id",
"join_type": "INNER"
},
{
"local_column": "category_id",
"foreign_table": "categories",
"foreign_column": "id",
"join_type": "INNER"
}
],
"order_by": [
{"column": "total_quantity", "direction": "DESC"}
],
"group_by": ["product_id", "category_id"],
"frequency": 25,
"avg_execution_time_ms": 180.5
},
{
"id": "customer_purchase_history",
"type": "SELECT",
"table": "orders",
"description": "Customer analytics - purchase history with items",
"where_conditions": [
{
"column": "user_id",
"operator": "=",
"selectivity": 0.8
},
{
"column": "status",
"operator": "IN",
"selectivity": 0.6
}
],
"join_conditions": [
{
"local_column": "id",
"foreign_table": "order_items",
"foreign_column": "order_id",
"join_type": "INNER"
}
],
"order_by": [
{"column": "created_at", "direction": "DESC"}
],
"group_by": [],
"frequency": 300,
"avg_execution_time_ms": 65.2
},
{
"id": "daily_sales_report",
"type": "SELECT",
"table": "orders",
"description": "Daily sales aggregation report",
"where_conditions": [
{
"column": "created_at",
"operator": ">=",
"selectivity": 0.05
},
{
"column": "status",
"operator": "IN",
"selectivity": 0.6
}
],
"join_conditions": [],
"order_by": [
{"column": "order_date", "direction": "DESC"}
],
"group_by": ["DATE(created_at)"],
"frequency": 10,
"avg_execution_time_ms": 250.8
},
{
"id": "category_hierarchy_nav",
"type": "SELECT",
"table": "categories",
"description": "Category navigation - parent-child relationships",
"where_conditions": [
{
"column": "parent_id",
"operator": "=",
"selectivity": 0.2
},
{
"column": "is_active",
"operator": "=",
"selectivity": 0.1
}
],
"join_conditions": [],
"order_by": [
{"column": "sort_order", "direction": "ASC"}
],
"group_by": [],
"frequency": 800,
"avg_execution_time_ms": 5.1
},
{
"id": "recent_user_reviews",
"type": "SELECT",
"table": "product_reviews",
"description": "Recent product reviews by user",
"where_conditions": [
{
"column": "user_id",
"operator": "=",
"selectivity": 0.95
}
],
"join_conditions": [
{
"local_column": "product_id",
"foreign_table": "products",
"foreign_column": "id",
"join_type": "INNER"
}
],
"order_by": [
{"column": "created_at", "direction": "DESC"}
],
"group_by": [],
"frequency": 200,
"avg_execution_time_ms": 12.7
},
{
"id": "product_avg_rating",
"type": "SELECT",
"table": "product_reviews",
"description": "Product average rating calculation",
"where_conditions": [
{
"column": "product_id",
"operator": "IN",
"selectivity": 0.1
}
],
"join_conditions": [],
"order_by": [],
"group_by": ["product_id"],
"frequency": 400,
"avg_execution_time_ms": 35.4
}
]
}{
"tables": {
"users": {
"columns": {
"id": {
"type": "INTEGER",
"nullable": false,
"unique": true,
"cardinality_estimate": 50000
},
"email": {
"type": "VARCHAR(255)",
"nullable": false,
"unique": true,
"cardinality_estimate": 50000
},
"username": {
"type": "VARCHAR(50)",
"nullable": false,
"unique": true,
"cardinality_estimate": 50000
},
"password_hash": {
"type": "VARCHAR(255)",
"nullable": false,
"cardinality_estimate": 50000
},
"first_name": {
"type": "VARCHAR(100)",
"nullable": true,
"cardinality_estimate": 25000
},
"last_name": {
"type": "VARCHAR(100)",
"nullable": true,
"cardinality_estimate": 30000
},
"status": {
"type": "VARCHAR(20)",
"nullable": false,
"default": "active",
"cardinality_estimate": 5
},
"created_at": {
"type": "TIMESTAMP",
"nullable": false,
"default": "CURRENT_TIMESTAMP"
}
},
"primary_key": ["id"],
"unique_constraints": [
["email"],
["username"]
],
"check_constraints": {
"chk_status_valid": "status IN ('active', 'inactive', 'suspended', 'deleted')"
},
"indexes": [
{
"name": "idx_users_email",
"columns": ["email"],
"unique": true
},
{
"name": "idx_users_status",
"columns": ["status"]
}
]
},
"products": {
"columns": {
"id": {
"type": "INTEGER",
"nullable": false,
"unique": true,
"cardinality_estimate": 10000
},
"name": {
"type": "VARCHAR(255)",
"nullable": false,
"cardinality_estimate": 9500
},
"sku": {
"type": "VARCHAR(50)",
"nullable": false,
"unique": true,
"cardinality_estimate": 10000
},
"price": {
"type": "DECIMAL(10,2)",
"nullable": false,
"cardinality_estimate": 5000
},
"category_id": {
"type": "INTEGER",
"nullable": false,
"foreign_key": "categories.id",
"cardinality_estimate": 50
},
"brand": {
"type": "VARCHAR(100)",
"nullable": true,
"cardinality_estimate": 200
},
"is_active": {
"type": "BOOLEAN",
"nullable": false,
"default": true,
"cardinality_estimate": 2
},
"inventory_count": {
"type": "INTEGER",
"nullable": false,
"default": 0,
"cardinality_estimate": 1000
},
"created_at": {
"type": "TIMESTAMP",
"nullable": false,
"default": "CURRENT_TIMESTAMP"
}
},
"primary_key": ["id"],
"unique_constraints": [
["sku"]
],
"check_constraints": {
"chk_price_positive": "price > 0",
"chk_inventory_non_negative": "inventory_count >= 0"
},
"indexes": [
{
"name": "idx_products_category",
"columns": ["category_id"]
},
{
"name": "idx_products_brand",
"columns": ["brand"]
},
{
"name": "idx_products_price",
"columns": ["price"]
},
{
"name": "idx_products_active_category",
"columns": ["is_active", "category_id"],
"partial_condition": "is_active = true"
}
]
},
"orders": {
"columns": {
"id": {
"type": "INTEGER",
"nullable": false,
"unique": true,
"cardinality_estimate": 200000
},
"order_number": {
"type": "VARCHAR(50)",
"nullable": false,
"unique": true,
"cardinality_estimate": 200000
},
"user_id": {
"type": "INTEGER",
"nullable": false,
"foreign_key": "users.id",
"cardinality_estimate": 40000
},
"status": {
"type": "VARCHAR(50)",
"nullable": false,
"default": "pending",
"cardinality_estimate": 8
},
"total_amount": {
"type": "DECIMAL(10,2)",
"nullable": false,
"cardinality_estimate": 50000
},
"payment_method": {
"type": "VARCHAR(50)",
"nullable": true,
"cardinality_estimate": 10
},
"created_at": {
"type": "TIMESTAMP",
"nullable": false,
"default": "CURRENT_TIMESTAMP"
},
"shipped_at": {
"type": "TIMESTAMP",
"nullable": true
}
},
"primary_key": ["id"],
"unique_constraints": [
["order_number"]
],
"check_constraints": {
"chk_total_positive": "total_amount > 0",
"chk_status_valid": "status IN ('pending', 'processing', 'shipped', 'delivered', 'cancelled')"
},
"indexes": [
{
"name": "idx_orders_user",
"columns": ["user_id"]
},
{
"name": "idx_orders_status",
"columns": ["status"]
},
{
"name": "idx_orders_created",
"columns": ["created_at"]
},
{
"name": "idx_orders_user_status",
"columns": ["user_id", "status"]
}
]
},
"order_items": {
"columns": {
"id": {
"type": "INTEGER",
"nullable": false,
"unique": true,
"cardinality_estimate": 800000
},
"order_id": {
"type": "INTEGER",
"nullable": false,
"foreign_key": "orders.id",
"cardinality_estimate": 200000
},
"product_id": {
"type": "INTEGER",
"nullable": false,
"foreign_key": "products.id",
"cardinality_estimate": 8000
},
"quantity": {
"type": "INTEGER",
"nullable": false,
"cardinality_estimate": 20
},
"unit_price": {
"type": "DECIMAL(10,2)",
"nullable": false,
"cardinality_estimate": 5000
},
"total_price": {
"type": "DECIMAL(10,2)",
"nullable": false,
"cardinality_estimate": 10000
}
},
"primary_key": ["id"],
"check_constraints": {
"chk_quantity_positive": "quantity > 0",
"chk_unit_price_positive": "unit_price > 0"
},
"indexes": [
{
"name": "idx_order_items_order",
"columns": ["order_id"]
},
{
"name": "idx_order_items_product",
"columns": ["product_id"]
}
]
},
"categories": {
"columns": {
"id": {
"type": "INTEGER",
"nullable": false,
"unique": true,
"cardinality_estimate": 100
},
"name": {
"type": "VARCHAR(100)",
"nullable": false,
"cardinality_estimate": 100
},
"parent_id": {
"type": "INTEGER",
"nullable": true,
"foreign_key": "categories.id",
"cardinality_estimate": 20
},
"is_active": {
"type": "BOOLEAN",
"nullable": false,
"default": true,
"cardinality_estimate": 2
}
},
"primary_key": ["id"],
"indexes": [
{
"name": "idx_categories_parent",
"columns": ["parent_id"]
},
{
"name": "idx_categories_active",
"columns": ["is_active"]
}
]
},
"product_reviews": {
"columns": {
"id": {
"type": "INTEGER",
"nullable": false,
"unique": true,
"cardinality_estimate": 150000
},
"product_id": {
"type": "INTEGER",
"nullable": false,
"foreign_key": "products.id",
"cardinality_estimate": 8000
},
"user_id": {
"type": "INTEGER",
"nullable": false,
"foreign_key": "users.id",
"cardinality_estimate": 30000
},
"rating": {
"type": "INTEGER",
"nullable": false,
"cardinality_estimate": 5
},
"review_text": {
"type": "TEXT",
"nullable": true
},
"created_at": {
"type": "TIMESTAMP",
"nullable": false,
"default": "CURRENT_TIMESTAMP"
}
},
"primary_key": ["id"],
"unique_constraints": [
["product_id", "user_id"]
],
"check_constraints": {
"chk_rating_valid": "rating BETWEEN 1 AND 5"
},
"indexes": [
{
"name": "idx_reviews_product",
"columns": ["product_id"]
},
{
"name": "idx_reviews_user",
"columns": ["user_id"]
},
{
"name": "idx_reviews_rating",
"columns": ["rating"]
}
]
}
}
}-- Sample E-commerce Database Schema
-- Demonstrates various normalization levels and common patterns
-- Users table - well normalized
CREATE TABLE users (
id INTEGER PRIMARY KEY,
email VARCHAR(255) NOT NULL UNIQUE,
username VARCHAR(50) NOT NULL UNIQUE,
password_hash VARCHAR(255) NOT NULL,
first_name VARCHAR(100),
last_name VARCHAR(100),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
status VARCHAR(20) DEFAULT 'active'
);
-- Categories table - hierarchical structure
CREATE TABLE categories (
id INTEGER PRIMARY KEY,
name VARCHAR(100) NOT NULL,
slug VARCHAR(100) NOT NULL UNIQUE,
parent_id INTEGER REFERENCES categories(id),
description TEXT,
is_active BOOLEAN DEFAULT true,
sort_order INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- Products table - potential normalization issues
CREATE TABLE products (
id INTEGER PRIMARY KEY,
name VARCHAR(255) NOT NULL,
sku VARCHAR(50) NOT NULL UNIQUE,
description TEXT,
price DECIMAL(10,2) NOT NULL,
cost DECIMAL(10,2),
weight DECIMAL(8,2),
dimensions VARCHAR(50), -- Potential 1NF violation: "10x5x3 inches"
category_id INTEGER REFERENCES categories(id),
category_name VARCHAR(100), -- Redundant with categories.name (3NF violation)
brand VARCHAR(100), -- Should be normalized to separate brands table
tags VARCHAR(500), -- Potential 1NF violation: comma-separated tags
inventory_count INTEGER DEFAULT 0,
reorder_point INTEGER DEFAULT 10,
supplier_name VARCHAR(100), -- Should be normalized
supplier_contact VARCHAR(255), -- Should be normalized
is_active BOOLEAN DEFAULT true,
featured BOOLEAN DEFAULT false,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- Addresses table - good normalization
CREATE TABLE addresses (
id INTEGER PRIMARY KEY,
user_id INTEGER REFERENCES users(id),
address_type VARCHAR(20) DEFAULT 'shipping', -- 'shipping', 'billing'
street_address VARCHAR(255) NOT NULL,
street_address_2 VARCHAR(255),
city VARCHAR(100) NOT NULL,
state VARCHAR(50) NOT NULL,
postal_code VARCHAR(20) NOT NULL,
country VARCHAR(50) NOT NULL DEFAULT 'US',
is_default BOOLEAN DEFAULT false,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- Orders table - mixed normalization issues
CREATE TABLE orders (
id INTEGER PRIMARY KEY,
order_number VARCHAR(50) NOT NULL UNIQUE,
user_id INTEGER REFERENCES users(id),
user_email VARCHAR(255), -- Denormalized for performance/historical reasons
user_name VARCHAR(200), -- Denormalized for performance/historical reasons
status VARCHAR(50) NOT NULL DEFAULT 'pending',
total_amount DECIMAL(10,2) NOT NULL,
tax_amount DECIMAL(10,2) NOT NULL,
shipping_amount DECIMAL(10,2) NOT NULL,
discount_amount DECIMAL(10,2) DEFAULT 0,
payment_method VARCHAR(50), -- Should be normalized to payment_methods
payment_status VARCHAR(50) DEFAULT 'pending',
shipping_address_id INTEGER REFERENCES addresses(id),
billing_address_id INTEGER REFERENCES addresses(id),
-- Denormalized shipping address for historical preservation
shipping_street VARCHAR(255),
shipping_city VARCHAR(100),
shipping_state VARCHAR(50),
shipping_postal_code VARCHAR(20),
shipping_country VARCHAR(50),
notes TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
shipped_at TIMESTAMP,
delivered_at TIMESTAMP
);
-- Order items table - properly normalized
CREATE TABLE order_items (
id INTEGER PRIMARY KEY,
order_id INTEGER REFERENCES orders(id),
product_id INTEGER REFERENCES products(id),
product_name VARCHAR(255), -- Denormalized for historical reasons
product_sku VARCHAR(50), -- Denormalized for historical reasons
quantity INTEGER NOT NULL,
unit_price DECIMAL(10,2) NOT NULL,
total_price DECIMAL(10,2) NOT NULL, -- Calculated field (could be computed)
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- Shopping cart table - session-based data
CREATE TABLE shopping_cart (
id INTEGER PRIMARY KEY,
user_id INTEGER REFERENCES users(id),
session_id VARCHAR(255), -- For anonymous users
product_id INTEGER REFERENCES products(id),
quantity INTEGER NOT NULL DEFAULT 1,
added_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(user_id, product_id),
UNIQUE(session_id, product_id)
);
-- Product reviews - user-generated content
CREATE TABLE product_reviews (
id INTEGER PRIMARY KEY,
product_id INTEGER REFERENCES products(id),
user_id INTEGER REFERENCES users(id),
rating INTEGER NOT NULL CHECK (rating BETWEEN 1 AND 5),
title VARCHAR(200),
review_text TEXT,
verified_purchase BOOLEAN DEFAULT false,
helpful_count INTEGER DEFAULT 0,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
UNIQUE(product_id, user_id) -- One review per user per product
);
-- Coupons table - promotional data
CREATE TABLE coupons (
id INTEGER PRIMARY KEY,
code VARCHAR(50) NOT NULL UNIQUE,
description VARCHAR(255),
discount_type VARCHAR(20) NOT NULL, -- 'percentage', 'fixed_amount'
discount_value DECIMAL(8,2) NOT NULL,
minimum_amount DECIMAL(10,2),
maximum_discount DECIMAL(10,2),
usage_limit INTEGER,
usage_count INTEGER DEFAULT 0,
valid_from TIMESTAMP NOT NULL,
valid_until TIMESTAMP NOT NULL,
is_active BOOLEAN DEFAULT true,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- Audit log table - for tracking changes
CREATE TABLE audit_log (
id INTEGER PRIMARY KEY,
table_name VARCHAR(50) NOT NULL,
record_id INTEGER NOT NULL,
action VARCHAR(20) NOT NULL, -- 'INSERT', 'UPDATE', 'DELETE'
old_values TEXT, -- JSON format
new_values TEXT, -- JSON format
user_id INTEGER REFERENCES users(id),
ip_address VARCHAR(45),
user_agent TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- Problematic table - multiple normalization violations
CREATE TABLE user_preferences (
user_id INTEGER PRIMARY KEY REFERENCES users(id),
preferred_categories VARCHAR(500), -- CSV list - 1NF violation
email_notifications VARCHAR(255), -- "daily,weekly,promotions" - 1NF violation
user_name VARCHAR(200), -- Redundant with users table - 3NF violation
user_email VARCHAR(255), -- Redundant with users table - 3NF violation
theme VARCHAR(50) DEFAULT 'light',
language VARCHAR(10) DEFAULT 'en',
timezone VARCHAR(50) DEFAULT 'UTC',
currency VARCHAR(3) DEFAULT 'USD',
date_format VARCHAR(20) DEFAULT 'YYYY-MM-DD',
newsletter_subscribed BOOLEAN DEFAULT true,
sms_notifications BOOLEAN DEFAULT false,
push_notifications BOOLEAN DEFAULT true,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
-- Create some basic indexes (some missing, some redundant for demonstration)
CREATE INDEX idx_users_email ON users (email);
CREATE INDEX idx_users_username ON users (username); -- Redundant due to UNIQUE constraint
CREATE INDEX idx_products_category ON products (category_id);
CREATE INDEX idx_products_brand ON products (brand);
CREATE INDEX idx_products_sku ON products (sku); -- Redundant due to UNIQUE constraint
CREATE INDEX idx_orders_user ON orders (user_id);
CREATE INDEX idx_orders_status ON orders (status);
CREATE INDEX idx_orders_created ON orders (created_at);
CREATE INDEX idx_order_items_order ON order_items (order_id);
CREATE INDEX idx_order_items_product ON order_items (product_id);
-- Missing index on addresses.user_id
-- Missing composite index on orders (user_id, status)
-- Missing index on product_reviews.product_id
-- Constraints that should exist but are missing
-- ALTER TABLE products ADD CONSTRAINT chk_price_positive CHECK (price > 0);
-- ALTER TABLE products ADD CONSTRAINT chk_inventory_non_negative CHECK (inventory_count >= 0);
-- ALTER TABLE order_items ADD CONSTRAINT chk_quantity_positive CHECK (quantity > 0);
-- ALTER TABLE orders ADD CONSTRAINT chk_total_positive CHECK (total_amount > 0);DATABASE INDEX OPTIMIZATION REPORT
==================================================
ANALYSIS SUMMARY
----------------
Tables Analyzed: 6
Query Patterns: 15
Existing Indexes: 12
New Recommendations: 8
High Priority: 4
Redundancy Issues: 2
HIGH PRIORITY RECOMMENDATIONS (4)
----------------------------------
1. orders: Optimize multi-column WHERE conditions: user_id, status, created_at
Columns: user_id, status, created_at
Benefit: Very High
SQL: CREATE INDEX idx_orders_user_status_created ON orders (user_id, status, created_at);
2. products: Optimize WHERE category_id = AND is_active = queries
Columns: category_id, is_active
Benefit: High
SQL: CREATE INDEX idx_products_category_active ON products (category_id, is_active);
3. order_items: Optimize JOIN with products table on product_id
Columns: product_id
Benefit: High (frequent JOINs)
SQL: CREATE INDEX idx_order_items_product_join ON order_items (product_id);
4. product_reviews: Covering index for WHERE + ORDER BY optimization
Columns: product_id, created_at
Benefit: High (eliminates table lookups for SELECT)
SQL: CREATE INDEX idx_product_reviews_covering_product_created ON product_reviews (product_id, created_at) INCLUDE (rating, review_text);
REDUNDANCY ISSUES (2)
---------------------
• DUPLICATE: Indexes 'idx_users_email' and 'unique_users_email' are identical
Recommendation: Drop one of the duplicate indexes
SQL: DROP INDEX idx_users_email;
• OVERLAPPING: Index 'idx_products_category' overlaps 85% with 'idx_products_category_active'
Recommendation: Consider dropping 'idx_products_category' as it's largely covered by 'idx_products_category_active'
SQL: DROP INDEX idx_products_category;
PERFORMANCE IMPACT ANALYSIS
----------------------------
Queries to be optimized: 12
High impact optimizations: 6
Estimated insert overhead: 40%
RECOMMENDED CREATE INDEX STATEMENTS
------------------------------------
1. CREATE INDEX idx_orders_user_status_created ON orders (user_id, status, created_at);
2. CREATE INDEX idx_products_category_active ON products (category_id, is_active);
3. CREATE INDEX idx_order_items_product_join ON order_items (product_id);
4. CREATE INDEX idx_product_reviews_covering_product_created ON product_reviews (product_id, created_at) INCLUDE (rating, review_text);
5. CREATE INDEX idx_products_price_brand ON products (price, brand);
6. CREATE INDEX idx_orders_status_created ON orders (status, created_at);
7. CREATE INDEX idx_categories_parent_active ON categories (parent_id, is_active);
8. CREATE INDEX idx_product_reviews_user_created ON product_reviews (user_id, created_at);DATABASE MIGRATION PLAN
==================================================
Migration ID: a7b3c9d2
Created: 2024-02-16T15:30:00Z
Zero Downtime: false
MIGRATION SUMMARY
-----------------
Total Steps: 12
Tables Added: 1
Tables Dropped: 0
Tables Renamed: 0
Columns Added: 3
Columns Dropped: 1
Columns Modified: 2
Constraints Added: 4
Constraints Dropped: 1
Indexes Added: 2
Indexes Dropped: 1
RISK ASSESSMENT
---------------
High Risk Steps: 3
Medium Risk Steps: 4
Low Risk Steps: 5
MIGRATION STEPS
---------------
1. Create table brands with 4 columns (LOW risk)
Type: CREATE_TABLE
Forward SQL: CREATE TABLE brands (
id INTEGER PRIMARY KEY,
name VARCHAR(100) NOT NULL,
description TEXT,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
Rollback SQL: DROP TABLE IF EXISTS brands;
2. Add column brand_id to products (LOW risk)
Type: ADD_COLUMN
Forward SQL: ALTER TABLE products ADD COLUMN brand_id INTEGER;
Rollback SQL: ALTER TABLE products DROP COLUMN brand_id;
3. Add column email_verified to users (LOW risk)
Type: ADD_COLUMN
Forward SQL: ALTER TABLE users ADD COLUMN email_verified BOOLEAN DEFAULT false;
Rollback SQL: ALTER TABLE users DROP COLUMN email_verified;
4. Add column last_login to users (LOW risk)
Type: ADD_COLUMN
Forward SQL: ALTER TABLE users ADD COLUMN last_login TIMESTAMP;
Rollback SQL: ALTER TABLE users DROP COLUMN last_login;
5. Modify column price: type: DECIMAL(10,2) -> DECIMAL(12,2) (LOW risk)
Type: MODIFY_COLUMN
Forward SQL: ALTER TABLE products
ALTER COLUMN price TYPE DECIMAL(12,2);
Rollback SQL: ALTER TABLE products
ALTER COLUMN price TYPE DECIMAL(10,2);
6. Modify column inventory_count: nullable: true -> false (HIGH risk)
Type: MODIFY_COLUMN
Forward SQL: ALTER TABLE products
ALTER COLUMN inventory_count SET NOT NULL;
Rollback SQL: ALTER TABLE products
ALTER COLUMN inventory_count DROP NOT NULL;
7. Add primary key on id (MEDIUM risk)
Type: ADD_CONSTRAINT
Forward SQL: ALTER TABLE brands ADD CONSTRAINT pk_brands PRIMARY KEY (id);
Rollback SQL: ALTER TABLE brands DROP CONSTRAINT pk_brands;
8. Add foreign key constraint on brand_id (MEDIUM risk)
Type: ADD_CONSTRAINT
Forward SQL: ALTER TABLE products ADD CONSTRAINT fk_products_brand_id FOREIGN KEY (brand_id) REFERENCES brands(id);
Rollback SQL: ALTER TABLE products DROP CONSTRAINT fk_products_brand_id;
9. Add unique constraint on name (MEDIUM risk)
Type: ADD_CONSTRAINT
Forward SQL: ALTER TABLE brands ADD CONSTRAINT uq_brands_name UNIQUE (name);
Rollback SQL: ALTER TABLE brands DROP CONSTRAINT uq_brands_name;
10. Add check constraint: price > 0 (MEDIUM risk)
Type: ADD_CONSTRAINT
Forward SQL: ALTER TABLE products ADD CONSTRAINT chk_products_price_positive CHECK (price > 0);
Rollback SQL: ALTER TABLE products DROP CONSTRAINT chk_products_price_positive;
11. Create index idx_products_brand_id on (brand_id) (LOW risk)
Type: ADD_INDEX
Forward SQL: CREATE INDEX idx_products_brand_id ON products (brand_id);
Rollback SQL: DROP INDEX idx_products_brand_id;
Estimated Time: 1-5 minutes depending on table size
12. Create index idx_users_email_verified on (email_verified) (LOW risk)
Type: ADD_INDEX
Forward SQL: CREATE INDEX idx_users_email_verified ON users (email_verified);
Rollback SQL: DROP INDEX idx_users_email_verified;
Estimated Time: 1-5 minutes depending on table size
VALIDATION CHECKS
-----------------
• Verify table brands exists
SQL: SELECT COUNT(*) FROM information_schema.tables WHERE table_name = 'brands';
Expected: 1
• Verify column brand_id exists in products
SQL: SELECT COUNT(*) FROM information_schema.columns WHERE table_name = 'products' AND column_name = 'brand_id';
Expected: 1
• Verify column email_verified exists in users
SQL: SELECT COUNT(*) FROM information_schema.columns WHERE table_name = 'users' AND column_name = 'email_verified';
Expected: 1
• Verify column modification in products
SQL: SELECT data_type, is_nullable FROM information_schema.columns WHERE table_name = 'products' AND column_name = 'price';
Expected: 1
• Verify index idx_products_brand_id exists
SQL: SELECT COUNT(*) FROM information_schema.statistics WHERE index_name = 'idx_products_brand_id';
Expected: 1
• Verify index idx_users_email_verified exists
SQL: SELECT COUNT(*) FROM information_schema.statistics WHERE index_name = 'idx_users_email_verified';
Expected: 1DATABASE SCHEMA ANALYSIS REPORT
==================================================
SCHEMA OVERVIEW
---------------
Total Tables: 8
Total Columns: 52
Tables with Primary Keys: 8
Total Foreign Keys: 6
Total Indexes: 15
KEY RECOMMENDATIONS
------------------
1. Address 3 high-severity issues immediately
2. Add primary keys to tables:
3. Review 4 VARCHAR(255) columns for right-sizing
4. Consider adding 2 foreign key constraints for referential integrity
5. Review 8 normalization issues for schema optimization
NORMALIZATION ISSUES (8 total)
------------------------------
High: 2, Medium: 3, Low: 2, Warning: 1
• products: Column 'dimensions' appears to store delimited values
Suggestion: Create separate table for individual values with foreign key relationship
• products: Column 'tags' appears to store delimited values
Suggestion: Create separate table for individual values with foreign key relationship
• products: Columns ['category_name'] may have transitive dependency through 'category_id'
Suggestion: Consider creating separate 'category' table with these columns
• orders: Columns ['shipping_street', 'shipping_city', 'shipping_state', 'shipping_postal_code', 'shipping_country'] may have transitive dependency through 'shipping_address_id'
Suggestion: Consider creating separate 'shipping_address' table with these columns
• user_preferences: Column 'preferred_categories' appears to store delimited values
Suggestion: Create separate table for individual values with foreign key relationship
DATA TYPE ISSUES (4 total)
--------------------------
• products.dimensions: VARCHAR(255) antipattern
Current: VARCHAR(50) → Suggested: Appropriately sized VARCHAR or TEXT
Rationale: VARCHAR(255) is often used as default without considering actual data length requirements
• products.tags: VARCHAR(255) antipattern
Current: VARCHAR(500) → Suggested: Appropriately sized VARCHAR or TEXT
Rationale: VARCHAR(255) is often used as default without considering actual data length requirements
• user_preferences.preferred_categories: VARCHAR(255) antipattern
Current: VARCHAR(500) → Suggested: Appropriately sized VARCHAR or TEXT
Rationale: VARCHAR(255) is often used as default without considering actual data length requirements
• user_preferences.email_notifications: VARCHAR(255) antipattern
Current: VARCHAR(255) → Suggested: Appropriately sized VARCHAR or TEXT
Rationale: VARCHAR(255) is often used as default without considering actual data length requirements
CONSTRAINT ISSUES (12 total)
-----------------------------
High: 0, Medium: 4, Low: 8
• products: Column 'price' should validate positive values
Suggestion: Add CHECK constraint: price > 0
• products: Column 'inventory_count' should validate positive values
Suggestion: Add CHECK constraint: inventory_count > 0
• orders: Column 'total_amount' should validate positive values
Suggestion: Add CHECK constraint: total_amount > 0
• order_items: Column 'quantity' should validate positive values
Suggestion: Add CHECK constraint: quantity > 0
• order_items: Column 'unit_price' should validate positive values
Suggestion: Add CHECK constraint: unit_price > 0
MISSING INDEXES (3 total)
-------------------------
• addresses.user_id (foreign_key)
SQL: CREATE INDEX idx_addresses_user_id ON addresses (user_id);
• product_reviews.product_id (foreign_key)
SQL: CREATE INDEX idx_product_reviews_product_id ON product_reviews (product_id);
• shopping_cart.user_id (foreign_key)
SQL: CREATE INDEX idx_shopping_cart_user_id ON shopping_cart (user_id);
MERMAID ERD
===========
erDiagram
USERS {
INTEGER id "PK"
VARCHAR(255) email "NOT NULL"
VARCHAR(50) username "NOT NULL"
VARCHAR(255) password_hash "NOT NULL"
VARCHAR(100) first_name
VARCHAR(100) last_name
TIMESTAMP created_at
TIMESTAMP updated_at
VARCHAR(20) status
}
CATEGORIES {
INTEGER id "PK"
VARCHAR(100) name "NOT NULL"
VARCHAR(100) slug "NOT NULL UNIQUE"
INTEGER parent_id "FK"
TEXT description
BOOLEAN is_active
INTEGER sort_order
TIMESTAMP created_at
}
PRODUCTS {
INTEGER id "PK"
VARCHAR(255) name "NOT NULL"
VARCHAR(50) sku "NOT NULL UNIQUE"
TEXT description
DECIMAL(10,2) price "NOT NULL"
DECIMAL(10,2) cost
DECIMAL(8,2) weight
VARCHAR(50) dimensions
INTEGER category_id "FK"
VARCHAR(100) category_name
VARCHAR(100) brand
VARCHAR(500) tags
INTEGER inventory_count
INTEGER reorder_point
VARCHAR(100) supplier_name
VARCHAR(255) supplier_contact
BOOLEAN is_active
BOOLEAN featured
TIMESTAMP created_at
TIMESTAMP updated_at
}
ADDRESSES {
INTEGER id "PK"
INTEGER user_id "FK"
VARCHAR(20) address_type
VARCHAR(255) street_address "NOT NULL"
VARCHAR(255) street_address_2
VARCHAR(100) city "NOT NULL"
VARCHAR(50) state "NOT NULL"
VARCHAR(20) postal_code "NOT NULL"
VARCHAR(50) country "NOT NULL"
BOOLEAN is_default
TIMESTAMP created_at
}
ORDERS {
INTEGER id "PK"
VARCHAR(50) order_number "NOT NULL UNIQUE"
INTEGER user_id "FK"
VARCHAR(255) user_email
VARCHAR(200) user_name
VARCHAR(50) status "NOT NULL"
DECIMAL(10,2) total_amount "NOT NULL"
DECIMAL(10,2) tax_amount "NOT NULL"
DECIMAL(10,2) shipping_amount "NOT NULL"
DECIMAL(10,2) discount_amount
VARCHAR(50) payment_method
VARCHAR(50) payment_status
INTEGER shipping_address_id "FK"
INTEGER billing_address_id "FK"
VARCHAR(255) shipping_street
VARCHAR(100) shipping_city
VARCHAR(50) shipping_state
VARCHAR(20) shipping_postal_code
VARCHAR(50) shipping_country
TEXT notes
TIMESTAMP created_at
TIMESTAMP updated_at
TIMESTAMP shipped_at
TIMESTAMP delivered_at
}
ORDER_ITEMS {
INTEGER id "PK"
INTEGER order_id "FK"
INTEGER product_id "FK"
VARCHAR(255) product_name
VARCHAR(50) product_sku
INTEGER quantity "NOT NULL"
DECIMAL(10,2) unit_price "NOT NULL"
DECIMAL(10,2) total_price "NOT NULL"
TIMESTAMP created_at
}
SHOPPING_CART {
INTEGER id "PK"
INTEGER user_id "FK"
VARCHAR(255) session_id
INTEGER product_id "FK"
INTEGER quantity "NOT NULL"
TIMESTAMP added_at
TIMESTAMP updated_at
}
PRODUCT_REVIEWS {
INTEGER id "PK"
INTEGER product_id "FK"
INTEGER user_id "FK"
INTEGER rating "NOT NULL"
VARCHAR(200) title
TEXT review_text
BOOLEAN verified_purchase
INTEGER helpful_count
TIMESTAMP created_at
TIMESTAMP updated_at
}
CATEGORIES ||--o{ CATEGORIES : has
CATEGORIES ||--o{ PRODUCTS : has
USERS ||--o{ ADDRESSES : has
USERS ||--o{ ORDERS : has
USERS ||--o{ SHOPPING_CART : has
USERS ||--o{ PRODUCT_REVIEWS : has
ADDRESSES ||--o{ ORDERS : has
ORDERS ||--o{ ORDER_ITEMS : has
PRODUCTS ||--o{ ORDER_ITEMS : has
PRODUCTS ||--o{ SHOPPING_CART : has
PRODUCTS ||--o{ PRODUCT_REVIEWS : has#!/usr/bin/env python3
"""
Database Index Optimizer
Analyzes schema definitions and query patterns to recommend optimal indexes:
- Identifies missing indexes for common query patterns
- Detects redundant and overlapping indexes
- Suggests composite index column ordering
- Estimates selectivity and performance impact
- Generates CREATE INDEX statements with rationale
Input: Schema JSON + Query patterns JSON
Output: Index recommendations + CREATE INDEX SQL + before/after analysis
Usage:
python index_optimizer.py --schema schema.json --queries queries.json --output recommendations.json
python index_optimizer.py --schema schema.json --queries queries.json --format text
python index_optimizer.py --schema schema.json --queries queries.json --analyze-existing
"""
import argparse
import json
import re
import sys
from collections import defaultdict, namedtuple, Counter
from typing import Dict, List, Set, Tuple, Optional, Any
from dataclasses import dataclass, asdict
import hashlib
@dataclass
class Column:
name: str
data_type: str
nullable: bool = True
unique: bool = False
cardinality_estimate: Optional[int] = None
@dataclass
class Index:
name: str
table: str
columns: List[str]
unique: bool = False
index_type: str = "btree"
partial_condition: Optional[str] = None
include_columns: List[str] = None
size_estimate: Optional[int] = None
@dataclass
class QueryPattern:
query_id: str
query_type: str # SELECT, INSERT, UPDATE, DELETE
table: str
where_conditions: List[Dict[str, Any]]
join_conditions: List[Dict[str, Any]]
order_by: List[Dict[str, str]] # column, direction
group_by: List[str]
frequency: int = 1
avg_execution_time_ms: Optional[float] = None
@dataclass
class IndexRecommendation:
recommendation_id: str
table: str
recommended_index: Index
reason: str
query_patterns_helped: List[str]
estimated_benefit: str
estimated_overhead: str
priority: int # 1 = highest priority
sql_statement: str
selectivity_analysis: Dict[str, Any]
@dataclass
class RedundancyIssue:
issue_type: str # DUPLICATE, OVERLAPPING, UNUSED
affected_indexes: List[str]
table: str
description: str
recommendation: str
sql_statements: List[str]
class SelectivityEstimator:
"""Estimates column selectivity based on naming patterns and data types."""
def __init__(self):
# Selectivity patterns based on common column names and types
self.high_selectivity_patterns = [
r'.*_id$', r'^id$', r'uuid', r'guid', r'email', r'username', r'ssn',
r'account.*number', r'transaction.*id', r'reference.*number'
]
self.medium_selectivity_patterns = [
r'name$', r'title$', r'description$', r'address', r'phone', r'zip',
r'postal.*code', r'serial.*number', r'sku', r'product.*code'
]
self.low_selectivity_patterns = [
r'status$', r'type$', r'category', r'state$', r'flag$', r'active$',
r'enabled$', r'deleted$', r'visible$', r'gender$', r'priority$'
]
self.very_low_selectivity_patterns = [
r'is_.*', r'has_.*', r'can_.*', r'boolean', r'bool'
]
def estimate_selectivity(self, column: Column, table_size_estimate: int = 10000) -> float:
"""Estimate column selectivity (0.0 = all same values, 1.0 = all unique values)."""
column_name_lower = column.name.lower()
# Primary key or unique columns
if column.unique or column.name.lower() in ['id', 'uuid', 'guid']:
return 1.0
# Check cardinality estimate if available
if column.cardinality_estimate:
return min(column.cardinality_estimate / table_size_estimate, 1.0)
# Pattern-based estimation
for pattern in self.high_selectivity_patterns:
if re.search(pattern, column_name_lower):
return 0.9 # Very high selectivity
for pattern in self.medium_selectivity_patterns:
if re.search(pattern, column_name_lower):
return 0.7 # Good selectivity
for pattern in self.low_selectivity_patterns:
if re.search(pattern, column_name_lower):
return 0.2 # Poor selectivity
for pattern in self.very_low_selectivity_patterns:
if re.search(pattern, column_name_lower):
return 0.1 # Very poor selectivity
# Data type based estimation
data_type_upper = column.data_type.upper()
if data_type_upper.startswith('BOOL'):
return 0.1
elif data_type_upper.startswith(('TINYINT', 'SMALLINT')):
return 0.3
elif data_type_upper.startswith('INT'):
return 0.8
elif data_type_upper.startswith(('VARCHAR', 'TEXT')):
# Estimate based on column name
if 'name' in column_name_lower:
return 0.7
elif 'description' in column_name_lower or 'comment' in column_name_lower:
return 0.9
else:
return 0.6
# Default moderate selectivity
return 0.5
class IndexOptimizer:
def __init__(self):
self.tables: Dict[str, Dict[str, Column]] = {}
self.existing_indexes: Dict[str, List[Index]] = {}
self.query_patterns: List[QueryPattern] = []
self.selectivity_estimator = SelectivityEstimator()
# Configuration
self.max_composite_index_columns = 6
self.min_selectivity_for_index = 0.1
self.redundancy_overlap_threshold = 0.8
def load_schema(self, schema_data: Dict[str, Any]) -> None:
"""Load schema definition."""
if 'tables' not in schema_data:
raise ValueError("Schema must contain 'tables' key")
for table_name, table_def in schema_data['tables'].items():
self.tables[table_name] = {}
self.existing_indexes[table_name] = []
# Load columns
for col_name, col_def in table_def.get('columns', {}).items():
column = Column(
name=col_name,
data_type=col_def.get('type', 'VARCHAR(255)'),
nullable=col_def.get('nullable', True),
unique=col_def.get('unique', False),
cardinality_estimate=col_def.get('cardinality_estimate')
)
self.tables[table_name][col_name] = column
# Load existing indexes
for idx_def in table_def.get('indexes', []):
index = Index(
name=idx_def['name'],
table=table_name,
columns=idx_def['columns'],
unique=idx_def.get('unique', False),
index_type=idx_def.get('type', 'btree'),
partial_condition=idx_def.get('partial_condition'),
include_columns=idx_def.get('include_columns', [])
)
self.existing_indexes[table_name].append(index)
def load_query_patterns(self, query_data: Dict[str, Any]) -> None:
"""Load query patterns for analysis."""
if 'queries' not in query_data:
raise ValueError("Query data must contain 'queries' key")
for query_def in query_data['queries']:
pattern = QueryPattern(
query_id=query_def['id'],
query_type=query_def.get('type', 'SELECT').upper(),
table=query_def['table'],
where_conditions=query_def.get('where_conditions', []),
join_conditions=query_def.get('join_conditions', []),
order_by=query_def.get('order_by', []),
group_by=query_def.get('group_by', []),
frequency=query_def.get('frequency', 1),
avg_execution_time_ms=query_def.get('avg_execution_time_ms')
)
self.query_patterns.append(pattern)
def analyze_missing_indexes(self) -> List[IndexRecommendation]:
"""Identify missing indexes based on query patterns."""
recommendations = []
for pattern in self.query_patterns:
table_name = pattern.table
if table_name not in self.tables:
continue
# Analyze WHERE conditions for single-column indexes
for condition in pattern.where_conditions:
column = condition.get('column')
operator = condition.get('operator', '=')
if column and column in self.tables[table_name]:
if not self._has_covering_index(table_name, [column]):
recommendation = self._create_single_column_recommendation(
table_name, column, pattern, operator
)
if recommendation:
recommendations.append(recommendation)
# Analyze composite indexes for multi-column WHERE conditions
where_columns = [cond.get('column') for cond in pattern.where_conditions
if cond.get('column') and cond.get('column') in self.tables[table_name]]
if len(where_columns) > 1:
composite_recommendation = self._create_composite_recommendation(
table_name, where_columns, pattern
)
if composite_recommendation:
recommendations.append(composite_recommendation)
# Analyze covering indexes for SELECT with ORDER BY
if pattern.order_by and where_columns:
covering_recommendation = self._create_covering_index_recommendation(
table_name, where_columns, pattern
)
if covering_recommendation:
recommendations.append(covering_recommendation)
# Analyze JOIN conditions
for join_condition in pattern.join_conditions:
local_column = join_condition.get('local_column')
if local_column and local_column in self.tables[table_name]:
if not self._has_covering_index(table_name, [local_column]):
recommendation = self._create_join_index_recommendation(
table_name, local_column, pattern, join_condition
)
if recommendation:
recommendations.append(recommendation)
# Remove duplicates and prioritize
recommendations = self._deduplicate_recommendations(recommendations)
recommendations = self._prioritize_recommendations(recommendations)
return recommendations
def _has_covering_index(self, table_name: str, columns: List[str]) -> bool:
"""Check if existing indexes cover the specified columns."""
if table_name not in self.existing_indexes:
return False
for index in self.existing_indexes[table_name]:
# Check if index starts with required columns (prefix match for composite)
if len(index.columns) >= len(columns):
if index.columns[:len(columns)] == columns:
return True
return False
def _create_single_column_recommendation(
self,
table_name: str,
column: str,
pattern: QueryPattern,
operator: str
) -> Optional[IndexRecommendation]:
"""Create recommendation for single-column index."""
column_obj = self.tables[table_name][column]
selectivity = self.selectivity_estimator.estimate_selectivity(column_obj)
# Skip very low selectivity columns unless frequently used
if selectivity < self.min_selectivity_for_index and pattern.frequency < 100:
return None
index_name = f"idx_{table_name}_{column}"
index = Index(
name=index_name,
table=table_name,
columns=[column],
unique=column_obj.unique,
index_type="btree"
)
reason = f"Optimize WHERE {column} {operator} queries"
if pattern.frequency > 10:
reason += f" (used {pattern.frequency} times)"
return IndexRecommendation(
recommendation_id=self._generate_recommendation_id(table_name, [column]),
table=table_name,
recommended_index=index,
reason=reason,
query_patterns_helped=[pattern.query_id],
estimated_benefit=self._estimate_benefit(selectivity, pattern.frequency),
estimated_overhead="Low (single column)",
priority=self._calculate_priority(selectivity, pattern.frequency, 1),
sql_statement=f"CREATE INDEX {index_name} ON {table_name} ({column});",
selectivity_analysis={
"column_selectivity": selectivity,
"estimated_reduction": f"{int(selectivity * 100)}%"
}
)
def _create_composite_recommendation(
self,
table_name: str,
columns: List[str],
pattern: QueryPattern
) -> Optional[IndexRecommendation]:
"""Create recommendation for composite index."""
if len(columns) > self.max_composite_index_columns:
columns = columns[:self.max_composite_index_columns]
# Order columns by selectivity (most selective first)
column_selectivities = []
for col in columns:
col_obj = self.tables[table_name][col]
selectivity = self.selectivity_estimator.estimate_selectivity(col_obj)
column_selectivities.append((col, selectivity))
# Sort by selectivity descending
column_selectivities.sort(key=lambda x: x[1], reverse=True)
ordered_columns = [col for col, _ in column_selectivities]
# Calculate combined selectivity
combined_selectivity = min(sum(sel for _, sel in column_selectivities) / len(columns), 0.95)
index_name = f"idx_{table_name}_{'_'.join(ordered_columns)}"
if len(index_name) > 63: # PostgreSQL limit
index_name = f"idx_{table_name}_composite_{abs(hash('_'.join(ordered_columns))) % 10000}"
index = Index(
name=index_name,
table=table_name,
columns=ordered_columns,
index_type="btree"
)
reason = f"Optimize multi-column WHERE conditions: {', '.join(ordered_columns)}"
return IndexRecommendation(
recommendation_id=self._generate_recommendation_id(table_name, ordered_columns),
table=table_name,
recommended_index=index,
reason=reason,
query_patterns_helped=[pattern.query_id],
estimated_benefit=self._estimate_benefit(combined_selectivity, pattern.frequency),
estimated_overhead=f"Medium (composite index with {len(ordered_columns)} columns)",
priority=self._calculate_priority(combined_selectivity, pattern.frequency, len(ordered_columns)),
sql_statement=f"CREATE INDEX {index_name} ON {table_name} ({', '.join(ordered_columns)});",
selectivity_analysis={
"column_selectivities": {col: sel for col, sel in column_selectivities},
"combined_selectivity": combined_selectivity,
"column_order_rationale": "Ordered by selectivity (most selective first)"
}
)
def _create_covering_index_recommendation(
self,
table_name: str,
where_columns: List[str],
pattern: QueryPattern
) -> Optional[IndexRecommendation]:
"""Create recommendation for covering index."""
order_columns = [col['column'] for col in pattern.order_by if col['column'] in self.tables[table_name]]
# Combine WHERE and ORDER BY columns
index_columns = where_columns.copy()
include_columns = []
# Add ORDER BY columns to index columns
for col in order_columns:
if col not in index_columns:
index_columns.append(col)
# Limit index columns
if len(index_columns) > self.max_composite_index_columns:
include_columns = index_columns[self.max_composite_index_columns:]
index_columns = index_columns[:self.max_composite_index_columns]
index_name = f"idx_{table_name}_covering_{'_'.join(index_columns[:3])}"
if len(index_name) > 63:
index_name = f"idx_{table_name}_covering_{abs(hash('_'.join(index_columns))) % 10000}"
index = Index(
name=index_name,
table=table_name,
columns=index_columns,
include_columns=include_columns,
index_type="btree"
)
reason = f"Covering index for WHERE + ORDER BY optimization"
# Calculate selectivity for main columns
main_selectivity = 0.5 # Default for covering indexes
if where_columns:
selectivities = [
self.selectivity_estimator.estimate_selectivity(self.tables[table_name][col])
for col in where_columns[:2] # Consider first 2 columns
]
main_selectivity = max(selectivities)
sql_parts = [f"CREATE INDEX {index_name} ON {table_name} ({', '.join(index_columns)})"]
if include_columns:
sql_parts.append(f" INCLUDE ({', '.join(include_columns)})")
sql_statement = ''.join(sql_parts) + ";"
return IndexRecommendation(
recommendation_id=self._generate_recommendation_id(table_name, index_columns, "covering"),
table=table_name,
recommended_index=index,
reason=reason,
query_patterns_helped=[pattern.query_id],
estimated_benefit="High (eliminates table lookups for SELECT)",
estimated_overhead=f"High (covering index with {len(index_columns)} columns)",
priority=self._calculate_priority(main_selectivity, pattern.frequency, len(index_columns)),
sql_statement=sql_statement,
selectivity_analysis={
"main_columns_selectivity": main_selectivity,
"covering_benefit": "Eliminates table lookup for SELECT queries"
}
)
def _create_join_index_recommendation(
self,
table_name: str,
column: str,
pattern: QueryPattern,
join_condition: Dict[str, Any]
) -> Optional[IndexRecommendation]:
"""Create recommendation for JOIN optimization index."""
column_obj = self.tables[table_name][column]
selectivity = self.selectivity_estimator.estimate_selectivity(column_obj)
index_name = f"idx_{table_name}_{column}_join"
index = Index(
name=index_name,
table=table_name,
columns=[column],
index_type="btree"
)
foreign_table = join_condition.get('foreign_table', 'unknown')
reason = f"Optimize JOIN with {foreign_table} table on {column}"
return IndexRecommendation(
recommendation_id=self._generate_recommendation_id(table_name, [column], "join"),
table=table_name,
recommended_index=index,
reason=reason,
query_patterns_helped=[pattern.query_id],
estimated_benefit=self._estimate_join_benefit(pattern.frequency),
estimated_overhead="Low (single column for JOIN)",
priority=2, # JOINs are generally high priority
sql_statement=f"CREATE INDEX {index_name} ON {table_name} ({column});",
selectivity_analysis={
"column_selectivity": selectivity,
"join_optimization": True
}
)
def _generate_recommendation_id(self, table: str, columns: List[str], suffix: str = "") -> str:
"""Generate unique recommendation ID."""
content = f"{table}_{'_'.join(sorted(columns))}_{suffix}"
return hashlib.md5(content.encode()).hexdigest()[:8]
def _estimate_benefit(self, selectivity: float, frequency: int) -> str:
"""Estimate performance benefit of index."""
if selectivity > 0.8 and frequency > 50:
return "Very High"
elif selectivity > 0.6 and frequency > 20:
return "High"
elif selectivity > 0.4 or frequency > 10:
return "Medium"
else:
return "Low"
def _estimate_join_benefit(self, frequency: int) -> str:
"""Estimate benefit for JOIN indexes."""
if frequency > 50:
return "Very High (frequent JOINs)"
elif frequency > 20:
return "High (regular JOINs)"
elif frequency > 5:
return "Medium (occasional JOINs)"
else:
return "Low (rare JOINs)"
def _calculate_priority(self, selectivity: float, frequency: int, column_count: int) -> int:
"""Calculate priority score (1 = highest priority)."""
# Base score calculation
score = 0
# Selectivity contribution (0-50 points)
score += int(selectivity * 50)
# Frequency contribution (0-30 points)
score += min(frequency, 30)
# Penalty for complex indexes (subtract points)
score -= (column_count - 1) * 5
# Convert to priority levels
if score >= 70:
return 1 # Highest
elif score >= 50:
return 2 # High
elif score >= 30:
return 3 # Medium
else:
return 4 # Low
def _deduplicate_recommendations(self, recommendations: List[IndexRecommendation]) -> List[IndexRecommendation]:
"""Remove duplicate recommendations."""
seen_indexes = set()
unique_recommendations = []
for rec in recommendations:
index_signature = (rec.table, tuple(rec.recommended_index.columns))
if index_signature not in seen_indexes:
seen_indexes.add(index_signature)
unique_recommendations.append(rec)
else:
# Merge query patterns helped
for existing_rec in unique_recommendations:
if (existing_rec.table == rec.table and
existing_rec.recommended_index.columns == rec.recommended_index.columns):
existing_rec.query_patterns_helped.extend(rec.query_patterns_helped)
break
return unique_recommendations
def _prioritize_recommendations(self, recommendations: List[IndexRecommendation]) -> List[IndexRecommendation]:
"""Sort recommendations by priority."""
return sorted(recommendations, key=lambda x: (x.priority, -len(x.query_patterns_helped)))
def analyze_redundant_indexes(self) -> List[RedundancyIssue]:
"""Identify redundant, overlapping, and potentially unused indexes."""
redundancy_issues = []
for table_name, indexes in self.existing_indexes.items():
if len(indexes) < 2:
continue
# Find duplicate indexes
duplicates = self._find_duplicate_indexes(table_name, indexes)
redundancy_issues.extend(duplicates)
# Find overlapping indexes
overlapping = self._find_overlapping_indexes(table_name, indexes)
redundancy_issues.extend(overlapping)
# Find potentially unused indexes
unused = self._find_unused_indexes(table_name, indexes)
redundancy_issues.extend(unused)
return redundancy_issues
def _find_duplicate_indexes(self, table_name: str, indexes: List[Index]) -> List[RedundancyIssue]:
"""Find exactly duplicate indexes."""
issues = []
seen_signatures = {}
for index in indexes:
signature = (tuple(index.columns), index.unique, index.partial_condition)
if signature in seen_signatures:
existing_index = seen_signatures[signature]
issues.append(RedundancyIssue(
issue_type="DUPLICATE",
affected_indexes=[existing_index.name, index.name],
table=table_name,
description=f"Indexes '{existing_index.name}' and '{index.name}' are identical",
recommendation=f"Drop one of the duplicate indexes",
sql_statements=[f"DROP INDEX {index.name};"]
))
else:
seen_signatures[signature] = index
return issues
def _find_overlapping_indexes(self, table_name: str, indexes: List[Index]) -> List[RedundancyIssue]:
"""Find overlapping indexes that might be redundant."""
issues = []
for i, index1 in enumerate(indexes):
for index2 in indexes[i+1:]:
overlap_ratio = self._calculate_overlap_ratio(index1, index2)
if overlap_ratio >= self.redundancy_overlap_threshold:
# Determine which index to keep
if len(index1.columns) <= len(index2.columns):
redundant_index = index1
keep_index = index2
else:
redundant_index = index2
keep_index = index1
issues.append(RedundancyIssue(
issue_type="OVERLAPPING",
affected_indexes=[index1.name, index2.name],
table=table_name,
description=f"Index '{redundant_index.name}' overlaps {int(overlap_ratio * 100)}% "
f"with '{keep_index.name}'",
recommendation=f"Consider dropping '{redundant_index.name}' as it's largely "
f"covered by '{keep_index.name}'",
sql_statements=[f"DROP INDEX {redundant_index.name};"]
))
return issues
def _calculate_overlap_ratio(self, index1: Index, index2: Index) -> float:
"""Calculate overlap ratio between two indexes."""
cols1 = set(index1.columns)
cols2 = set(index2.columns)
if not cols1 or not cols2:
return 0.0
intersection = len(cols1.intersection(cols2))
union = len(cols1.union(cols2))
return intersection / union if union > 0 else 0.0
def _find_unused_indexes(self, table_name: str, indexes: List[Index]) -> List[RedundancyIssue]:
"""Find potentially unused indexes based on query patterns."""
issues = []
# Collect all columns used in query patterns for this table
used_columns = set()
table_patterns = [p for p in self.query_patterns if p.table == table_name]
for pattern in table_patterns:
# Add WHERE condition columns
for condition in pattern.where_conditions:
if condition.get('column'):
used_columns.add(condition['column'])
# Add JOIN columns
for join in pattern.join_conditions:
if join.get('local_column'):
used_columns.add(join['local_column'])
# Add ORDER BY columns
for order in pattern.order_by:
if order.get('column'):
used_columns.add(order['column'])
# Add GROUP BY columns
used_columns.update(pattern.group_by)
if not used_columns:
return issues # Can't determine usage without query patterns
for index in indexes:
index_columns = set(index.columns)
if not index_columns.intersection(used_columns):
issues.append(RedundancyIssue(
issue_type="UNUSED",
affected_indexes=[index.name],
table=table_name,
description=f"Index '{index.name}' columns {index.columns} are not used in any query patterns",
recommendation="Consider dropping this index if it's truly unused (verify with query logs)",
sql_statements=[f"-- Review usage before dropping\n-- DROP INDEX {index.name};"]
))
return issues
def estimate_index_sizes(self) -> Dict[str, Dict[str, Any]]:
"""Estimate storage requirements for recommended indexes."""
size_estimates = {}
# This is a simplified estimation - in practice, would need actual table statistics
for table_name in self.tables:
size_estimates[table_name] = {
"estimated_table_rows": 10000, # Default estimate
"existing_indexes_size_mb": len(self.existing_indexes.get(table_name, [])) * 5, # Rough estimate
"index_overhead_per_column_mb": 2 # Rough estimate per column
}
return size_estimates
def generate_analysis_report(self) -> Dict[str, Any]:
"""Generate comprehensive analysis report."""
recommendations = self.analyze_missing_indexes()
redundancy_issues = self.analyze_redundant_indexes()
size_estimates = self.estimate_index_sizes()
# Calculate statistics
total_existing_indexes = sum(len(indexes) for indexes in self.existing_indexes.values())
tables_analyzed = len(self.tables)
query_patterns_analyzed = len(self.query_patterns)
# Categorize recommendations by priority
high_priority = [r for r in recommendations if r.priority <= 2]
medium_priority = [r for r in recommendations if r.priority == 3]
low_priority = [r for r in recommendations if r.priority >= 4]
return {
"analysis_summary": {
"tables_analyzed": tables_analyzed,
"query_patterns_analyzed": query_patterns_analyzed,
"existing_indexes": total_existing_indexes,
"total_recommendations": len(recommendations),
"high_priority_recommendations": len(high_priority),
"redundancy_issues_found": len(redundancy_issues)
},
"index_recommendations": {
"high_priority": [asdict(r) for r in high_priority],
"medium_priority": [asdict(r) for r in medium_priority],
"low_priority": [asdict(r) for r in low_priority]
},
"redundancy_analysis": [asdict(issue) for issue in redundancy_issues],
"size_estimates": size_estimates,
"sql_statements": {
"create_indexes": [rec.sql_statement for rec in recommendations],
"drop_redundant": [
stmt for issue in redundancy_issues
for stmt in issue.sql_statements
]
},
"performance_impact": self._generate_performance_impact_analysis(recommendations)
}
def _generate_performance_impact_analysis(self, recommendations: List[IndexRecommendation]) -> Dict[str, Any]:
"""Generate performance impact analysis."""
impact_analysis = {
"query_optimization": {},
"write_overhead": {},
"storage_impact": {}
}
# Analyze query optimization impact
query_benefits = defaultdict(list)
for rec in recommendations:
for query_id in rec.query_patterns_helped:
query_benefits[query_id].append(rec.estimated_benefit)
impact_analysis["query_optimization"] = {
"queries_improved": len(query_benefits),
"high_impact_queries": len([q for q, benefits in query_benefits.items()
if any("High" in benefit for benefit in benefits)]),
"benefit_distribution": dict(Counter(
rec.estimated_benefit for rec in recommendations
))
}
# Analyze write overhead
impact_analysis["write_overhead"] = {
"total_new_indexes": len(recommendations),
"estimated_insert_overhead": f"{len(recommendations) * 5}%", # Rough estimate
"tables_most_affected": list(Counter(rec.table for rec in recommendations).most_common(3))
}
return impact_analysis
def format_text_report(self, analysis: Dict[str, Any]) -> str:
"""Format analysis as human-readable text report."""
lines = []
lines.append("DATABASE INDEX OPTIMIZATION REPORT")
lines.append("=" * 50)
lines.append("")
# Summary
summary = analysis["analysis_summary"]
lines.append("ANALYSIS SUMMARY")
lines.append("-" * 16)
lines.append(f"Tables Analyzed: {summary['tables_analyzed']}")
lines.append(f"Query Patterns: {summary['query_patterns_analyzed']}")
lines.append(f"Existing Indexes: {summary['existing_indexes']}")
lines.append(f"New Recommendations: {summary['total_recommendations']}")
lines.append(f"High Priority: {summary['high_priority_recommendations']}")
lines.append(f"Redundancy Issues: {summary['redundancy_issues_found']}")
lines.append("")
# High Priority Recommendations
high_priority = analysis["index_recommendations"]["high_priority"]
if high_priority:
lines.append(f"HIGH PRIORITY RECOMMENDATIONS ({len(high_priority)})")
lines.append("-" * 35)
for i, rec in enumerate(high_priority[:10], 1): # Show top 10
lines.append(f"{i}. {rec['table']}: {rec['reason']}")
lines.append(f" Columns: {', '.join(rec['recommended_index']['columns'])}")
lines.append(f" Benefit: {rec['estimated_benefit']}")
lines.append(f" SQL: {rec['sql_statement']}")
lines.append("")
# Redundancy Issues
redundancy = analysis["redundancy_analysis"]
if redundancy:
lines.append(f"REDUNDANCY ISSUES ({len(redundancy)})")
lines.append("-" * 20)
for issue in redundancy[:5]: # Show first 5
lines.append(f"• {issue['issue_type']}: {issue['description']}")
lines.append(f" Recommendation: {issue['recommendation']}")
if issue['sql_statements']:
lines.append(f" SQL: {issue['sql_statements'][0]}")
lines.append("")
# Performance Impact
perf_impact = analysis["performance_impact"]
lines.append("PERFORMANCE IMPACT ANALYSIS")
lines.append("-" * 30)
query_opt = perf_impact["query_optimization"]
lines.append(f"Queries to be optimized: {query_opt['queries_improved']}")
lines.append(f"High impact optimizations: {query_opt['high_impact_queries']}")
write_overhead = perf_impact["write_overhead"]
lines.append(f"Estimated insert overhead: {write_overhead['estimated_insert_overhead']}")
lines.append("")
# SQL Statements Summary
sql_statements = analysis["sql_statements"]
create_statements = sql_statements["create_indexes"]
if create_statements:
lines.append("RECOMMENDED CREATE INDEX STATEMENTS")
lines.append("-" * 36)
for i, stmt in enumerate(create_statements[:10], 1):
lines.append(f"{i}. {stmt}")
if len(create_statements) > 10:
lines.append(f"... and {len(create_statements) - 10} more")
lines.append("")
return "\n".join(lines)
def main():
parser = argparse.ArgumentParser(description="Optimize database indexes based on schema and query patterns")
parser.add_argument("--schema", "-s", required=True, help="Schema definition JSON file")
parser.add_argument("--queries", "-q", required=True, help="Query patterns JSON file")
parser.add_argument("--output", "-o", help="Output file (default: stdout)")
parser.add_argument("--format", "-f", choices=["json", "text"], default="text",
help="Output format")
parser.add_argument("--analyze-existing", "-e", action="store_true",
help="Include analysis of existing indexes")
parser.add_argument("--min-priority", "-p", type=int, default=4,
help="Minimum priority level to include (1=highest, 4=lowest)")
args = parser.parse_args()
try:
# Load schema
with open(args.schema, 'r') as f:
schema_data = json.load(f)
# Load queries
with open(args.queries, 'r') as f:
query_data = json.load(f)
# Initialize optimizer
optimizer = IndexOptimizer()
optimizer.load_schema(schema_data)
optimizer.load_query_patterns(query_data)
# Generate analysis
analysis = optimizer.generate_analysis_report()
# Filter by priority if specified
if args.min_priority < 4:
for priority_level in ["high_priority", "medium_priority", "low_priority"]:
analysis["index_recommendations"][priority_level] = [
rec for rec in analysis["index_recommendations"][priority_level]
if rec["priority"] <= args.min_priority
]
# Format output
if args.format == "json":
output = json.dumps(analysis, indent=2)
else:
output = optimizer.format_text_report(analysis)
# Write output
if args.output:
with open(args.output, 'w') as f:
f.write(output)
else:
print(output)
return 0
except Exception as e:
print(f"Error: {e}", file=sys.stderr)
return 1
if __name__ == "__main__":
sys.exit(main())Database Designer - POWERFUL Tier Skill
A comprehensive database design and analysis toolkit that provides expert-level schema analysis, index optimization, and migration generation capabilities for modern database systems.
Features
🔍 Schema Analyzer
- Normalization Analysis: Automated detection of 1NF through BCNF violations
- Data Type Optimization: Identifies antipatterns and inappropriate types
- Constraint Analysis: Finds missing foreign keys, unique constraints, and checks
- ERD Generation: Creates Mermaid diagrams from DDL or JSON schema
- Naming Convention Validation: Ensures consistent naming patterns
⚡ Index Optimizer
- Missing Index Detection: Identifies indexes needed for query patterns
- Composite Index Design: Optimizes column ordering for maximum efficiency
- Redundancy Analysis: Finds duplicate and overlapping indexes
- Performance Modeling: Estimates selectivity and query performance impact
- Covering Index Recommendations: Eliminates table lookups
🚀 Migration Generator
- Zero-Downtime Migrations: Implements expand-contract patterns
- Schema Evolution: Handles column changes, table renames, constraint updates
- Data Migration Scripts: Automated data transformation and validation
- Rollback Planning: Complete reversal capabilities for all changes
- Execution Orchestration: Dependency-aware migration ordering
Quick Start
Prerequisites
- Python 3.7+ (no external dependencies required)
- Database schema in SQL DDL format or JSON
- Query patterns (for index optimization)
Installation
# Clone or download the database-designer skill
cd engineering/database-designer/
# Make scripts executable
chmod +x *.pyUsage Examples
Schema Analysis
Analyze SQL DDL file:
python schema_analyzer.py --input assets/sample_schema.sql --output-format textGenerate ERD diagram:
python schema_analyzer.py --input assets/sample_schema.sql --generate-erd --output analysis.txtJSON schema analysis:
python schema_analyzer.py --input assets/sample_schema.json --output-format json --output results.jsonIndex Optimization
Basic index analysis:
python index_optimizer.py --schema assets/sample_schema.json --queries assets/sample_query_patterns.jsonHigh-priority recommendations only:
python index_optimizer.py --schema assets/sample_schema.json --queries assets/sample_query_patterns.json --min-priority 2JSON output with existing index analysis:
python index_optimizer.py --schema assets/sample_schema.json --queries assets/sample_query_patterns.json --format json --analyze-existingMigration Generation
Generate migration between schemas:
python migration_generator.py --current assets/current_schema.json --target assets/target_schema.jsonZero-downtime migration:
python migration_generator.py --current current.json --target target.json --zero-downtime --format sqlInclude validation queries:
python migration_generator.py --current current.json --target target.json --include-validations --output migration_plan.txtTool Documentation
Schema Analyzer
Input Formats:
- SQL DDL files (.sql)
- JSON schema definitions (.json)
Key Capabilities:
- Detects 1NF violations (non-atomic values, repeating groups)
- Identifies 2NF issues (partial dependencies in composite keys)
- Finds 3NF problems (transitive dependencies)
- Checks BCNF compliance (determinant key requirements)
- Validates data types (VARCHAR(255) antipattern, inappropriate types)
- Missing constraints (NOT NULL, UNIQUE, CHECK, foreign keys)
- Naming convention adherence
Sample Command:
python schema_analyzer.py \
--input sample_schema.sql \
--generate-erd \
--output-format text \
--output analysis.txtOutput:
- Comprehensive text or JSON analysis report
- Mermaid ERD diagram
- Prioritized recommendations
- SQL statements for improvements
Index Optimizer
Input Requirements:
- Schema definition (JSON format)
- Query patterns with frequency and selectivity data
Analysis Features:
- Selectivity estimation based on column patterns
- Composite index column ordering optimization
- Covering index recommendations for SELECT queries
- Foreign key index validation
- Redundancy detection (duplicates, overlaps, unused indexes)
- Performance impact modeling
Sample Command:
python index_optimizer.py \
--schema schema.json \
--queries query_patterns.json \
--format text \
--min-priority 3 \
--output recommendations.txtOutput:
- Prioritized index recommendations
- CREATE INDEX statements
- Drop statements for redundant indexes
- Performance impact analysis
- Storage size estimates
Migration Generator
Input Requirements:
- Current schema (JSON format)
- Target schema (JSON format)
Migration Strategies:
- Standard migrations with ALTER statements
- Zero-downtime expand-contract patterns
- Data migration and transformation scripts
- Constraint management (add/drop in correct order)
- Index management with timing estimates
Sample Command:
python migration_generator.py \
--current current_schema.json \
--target target_schema.json \
--zero-downtime \
--include-validations \
--format textOutput:
- Step-by-step migration plan
- Forward and rollback SQL statements
- Risk assessment for each step
- Validation queries
- Execution time estimates
File Structure
database-designer/
├── README.md # This file
├── SKILL.md # Comprehensive database design guide
├── schema_analyzer.py # Schema analysis tool
├── index_optimizer.py # Index optimization tool
├── migration_generator.py # Migration generation tool
├── references/ # Reference documentation
│ ├── normalization_guide.md # Normalization principles and patterns
│ ├── index_strategy_patterns.md # Index design and optimization guide
│ └── database_selection_decision_tree.md # Database technology selection
├── assets/ # Sample files and test data
│ ├── sample_schema.sql # Sample DDL with various issues
│ ├── sample_schema.json # JSON schema definition
│ └── sample_query_patterns.json # Query patterns for index analysis
└── expected_outputs/ # Example tool outputs
├── schema_analysis_sample.txt # Sample schema analysis report
├── index_optimization_sample.txt # Sample index recommendations
└── migration_sample.txt # Sample migration planJSON Schema Format
The tools use a standardized JSON format for schema definitions:
{
"tables": {
"table_name": {
"columns": {
"column_name": {
"type": "VARCHAR(255)",
"nullable": true,
"unique": false,
"foreign_key": "other_table.column",
"default": "default_value",
"cardinality_estimate": 1000
}
},
"primary_key": ["id"],
"unique_constraints": [["email"], ["username"]],
"check_constraints": {
"chk_positive_price": "price > 0"
},
"indexes": [
{
"name": "idx_table_column",
"columns": ["column_name"],
"unique": false,
"partial_condition": "status = 'active'"
}
]
}
}
}Query Patterns Format
For index optimization, provide query patterns in this format:
{
"queries": [
{
"id": "user_lookup",
"type": "SELECT",
"table": "users",
"where_conditions": [
{
"column": "email",
"operator": "=",
"selectivity": 0.95
}
],
"join_conditions": [
{
"local_column": "user_id",
"foreign_table": "orders",
"foreign_column": "id",
"join_type": "INNER"
}
],
"order_by": [
{"column": "created_at", "direction": "DESC"}
],
"frequency": 1000,
"avg_execution_time_ms": 5.2
}
]
}Best Practices
Schema Analysis
1. Start with DDL: Use actual CREATE TABLE statements when possible 2. Include Constraints: Capture all existing constraints and indexes 3. Consider History: Some denormalization may be intentional for performance 4. Validate Results: Review recommendations against business requirements
Index Optimization
1. Real Query Patterns: Use actual application queries, not theoretical ones 2. Include Frequency: Query frequency is crucial for prioritization 3. Monitor Performance: Validate recommendations with actual performance testing 4. Gradual Implementation: Add indexes incrementally and monitor impact
Migration Planning
1. Test Migrations: Always test on non-production environments first 2. Backup First: Ensure complete backups before running migrations 3. Monitor Progress: Watch for locks and performance impacts during execution 4. Rollback Ready: Have rollback procedures tested and ready
Advanced Usage
Custom Selectivity Estimation
The index optimizer uses pattern-based selectivity estimation. You can improve accuracy by providing cardinality estimates in your schema JSON:
{
"columns": {
"status": {
"type": "VARCHAR(20)",
"cardinality_estimate": 5 # Only 5 distinct values
}
}
}Zero-Downtime Migration Strategy
For production systems, use the zero-downtime flag to generate expand-contract migrations:
1. Expand Phase: Add new columns/tables without constraints 2. Dual Write: Application writes to both old and new structures 3. Backfill: Populate new structures with existing data 4. Contract Phase: Remove old structures after validation
Integration with CI/CD
Integrate these tools into your deployment pipeline:
# Schema validation in CI
python schema_analyzer.py --input schema.sql --output-format json | \
jq '.constraint_analysis.total_issues' | \
test $(cat) -eq 0 || exit 1
# Generate migrations automatically
python migration_generator.py \
--current prod_schema.json \
--target new_schema.json \
--zero-downtime \
--output migration.sqlTroubleshooting
Common Issues
"No tables found in input file"
- Ensure SQL DDL uses standard CREATE TABLE syntax
- Check for syntax errors in DDL
- Verify file encoding (UTF-8 recommended)
"Invalid JSON schema"
- Validate JSON syntax with a JSON validator
- Ensure all required fields are present
- Check that foreign key references use "table.column" format
"Analysis shows no issues but problems exist"
- Tools use heuristic analysis - review recommendations carefully
- Some design decisions may be intentional (denormalization for performance)
- Consider domain-specific requirements not captured by general rules
Performance Tips
Large Schemas:
- Use
--output-format jsonfor machine processing - Consider analyzing subsets of tables for very large schemas
- Provide cardinality estimates for better index recommendations
Complex Queries:
- Include actual execution times in query patterns
- Provide realistic frequency estimates
- Consider seasonal or usage pattern variations
Contributing
This is a self-contained skill with no external dependencies. To extend functionality:
1. Follow the existing code patterns 2. Maintain Python standard library only requirement 3. Add comprehensive test cases for new features 4. Update documentation and examples
License
This database designer skill is part of the claude-skills collection and follows the same licensing terms.
Database Selection Decision Tree
Overview
Choosing the right database technology is crucial for application success. This guide provides a systematic approach to database selection based on specific requirements, data patterns, and operational constraints.
Decision Framework
Primary Questions
1. What is your primary use case?
- OLTP (Online Transaction Processing)
- OLAP (Online Analytical Processing)
- Real-time analytics
- Content management
- Search and discovery
- Time-series data
- Graph relationships
2. What are your consistency requirements?
- Strong consistency (ACID)
- Eventual consistency
- Causal consistency
- Session consistency
3. What are your scalability needs?
- Vertical scaling sufficient
- Horizontal scaling required
- Global distribution needed
- Multi-region requirements
4. What is your data structure?
- Structured (relational)
- Semi-structured (JSON/XML)
- Unstructured (documents, media)
- Graph relationships
- Time-series data
- Key-value pairs
Decision Tree
START: What is your primary use case?
│
├── OLTP (Transactional Applications)
│ │
│ ├── Do you need strong ACID guarantees?
│ │ ├── YES → Do you need horizontal scaling?
│ │ │ ├── YES → Distributed SQL
│ │ │ │ ├── CockroachDB (Global, multi-region)
│ │ │ │ ├── TiDB (MySQL compatibility)
│ │ │ │ └── Spanner (Google Cloud)
│ │ │ └── NO → Traditional SQL
│ │ │ ├── PostgreSQL (Feature-rich, extensions)
│ │ │ ├── MySQL (Performance, ecosystem)
│ │ │ └── SQL Server (Microsoft stack)
│ │ └── NO → Are you primarily key-value access?
│ │ ├── YES → Key-Value Stores
│ │ │ ├── Redis (In-memory, caching)
│ │ │ ├── DynamoDB (AWS managed)
│ │ │ └── Cassandra (High availability)
│ │ └── NO → Document Stores
│ │ ├── MongoDB (General purpose)
│ │ ├── CouchDB (Sync, replication)
│ │ └── Amazon DocumentDB (MongoDB compatible)
│ │
├── OLAP (Analytics and Reporting)
│ │
│ ├── What is your data volume?
│ │ ├── Small to Medium (< 1TB) → Traditional SQL with optimization
│ │ │ ├── PostgreSQL with columnar extensions
│ │ │ ├── MySQL with analytics engine
│ │ │ └── SQL Server with columnstore
│ │ ├── Large (1TB - 100TB) → Data Warehouse Solutions
│ │ │ ├── Snowflake (Cloud-native)
│ │ │ ├── BigQuery (Google Cloud)
│ │ │ ├── Redshift (AWS)
│ │ │ └── Synapse (Azure)
│ │ └── Very Large (> 100TB) → Big Data Platforms
│ │ ├── Databricks (Unified analytics)
│ │ ├── Apache Spark on cloud
│ │ └── Hadoop ecosystem
│ │
├── Real-time Analytics
│ │
│ ├── Do you need sub-second query responses?
│ │ ├── YES → Stream Processing + OLAP
│ │ │ ├── ClickHouse (Fast analytics)
│ │ │ ├── Apache Druid (Real-time OLAP)
│ │ │ ├── Pinot (LinkedIn's real-time DB)
│ │ │ └── TimescaleDB (Time-series)
│ │ └── NO → Traditional OLAP solutions
│ │
├── Search and Discovery
│ │
│ ├── What type of search?
│ │ ├── Full-text search → Search Engines
│ │ │ ├── Elasticsearch (Full-featured)
│ │ │ ├── OpenSearch (AWS fork of ES)
│ │ │ └── Solr (Apache Lucene-based)
│ │ ├── Vector/similarity search → Vector Databases
│ │ │ ├── Pinecone (Managed vector DB)
│ │ │ ├── Weaviate (Open source)
│ │ │ ├── Chroma (Embeddings)
│ │ │ └── PostgreSQL with pgvector
│ │ └── Faceted search → Search + SQL combination
│ │
├── Graph Relationships
│ │
│ ├── Do you need complex graph traversals?
│ │ ├── YES → Graph Databases
│ │ │ ├── Neo4j (Property graph)
│ │ │ ├── Amazon Neptune (Multi-model)
│ │ │ ├── ArangoDB (Multi-model)
│ │ │ └── TigerGraph (Analytics focused)
│ │ └── NO → SQL with recursive queries
│ │ └── PostgreSQL with recursive CTEs
│ │
└── Time-series Data
│
├── What is your write volume?
├── High (millions/sec) → Specialized Time-series
│ ├── InfluxDB (Purpose-built)
│ ├── TimescaleDB (PostgreSQL extension)
│ ├── Apache Druid (Analytics focused)
│ └── Prometheus (Monitoring)
└── Medium → SQL with time-series optimization
└── PostgreSQL with partitioningDatabase Categories Deep Dive
Traditional SQL Databases
PostgreSQL
- Best For: Complex queries, JSON data, extensions, geospatial
- Strengths: Feature-rich, reliable, strong consistency, extensible
- Use Cases: OLTP, mixed workloads, JSON documents, geospatial applications
- Scaling: Vertical scaling, read replicas, partitioning
- When to Choose: Need SQL features, complex queries, moderate scale
MySQL
- Best For: Web applications, read-heavy workloads, simple schema
- Strengths: Performance, replication, large ecosystem
- Use Cases: Web apps, content management, e-commerce
- Scaling: Read replicas, sharding, clustering (MySQL Cluster)
- When to Choose: Simple schema, performance priority, large community
SQL Server
- Best For: Microsoft ecosystem, enterprise features, business intelligence
- Strengths: Integration, tooling, enterprise features
- Use Cases: Enterprise applications, .NET applications, BI
- Scaling: Always On availability groups, partitioning
- When to Choose: Microsoft stack, enterprise requirements
Distributed SQL (NewSQL)
CockroachDB
- Best For: Global applications, strong consistency, horizontal scaling
- Strengths: ACID guarantees, automatic scaling, survival
- Use Cases: Multi-region apps, financial services, global SaaS
- Trade-offs: Complex setup, higher latency for global transactions
- When to Choose: Need SQL + global scale + consistency
TiDB
- Best For: MySQL compatibility with horizontal scaling
- Strengths: MySQL protocol, HTAP (hybrid), cloud-native
- Use Cases: MySQL migrations, hybrid workloads
- When to Choose: Existing MySQL expertise, need scale
NoSQL Document Stores
MongoDB
- Best For: Flexible schema, rapid development, document-centric data
- Strengths: Developer experience, flexible schema, rich queries
- Use Cases: Content management, catalogs, user profiles, IoT
- Scaling: Automatic sharding, replica sets
- When to Choose: Schema evolution, document structure, rapid development
CouchDB
- Best For: Offline-first applications, multi-master replication
- Strengths: HTTP API, replication, conflict resolution
- Use Cases: Mobile apps, distributed systems, offline scenarios
- When to Choose: Need offline capabilities, bi-directional sync
Key-Value Stores
Redis
- Best For: Caching, sessions, real-time applications, pub/sub
- Strengths: Performance, data structures, persistence options
- Use Cases: Caching, leaderboards, real-time analytics, queues
- Scaling: Clustering, sentinel for HA
- When to Choose: High performance, simple data model, caching
DynamoDB
- Best For: Serverless applications, predictable performance, AWS ecosystem
- Strengths: Managed, auto-scaling, consistent performance
- Use Cases: Web applications, gaming, IoT, mobile backends
- Trade-offs: Vendor lock-in, limited querying
- When to Choose: AWS ecosystem, serverless, managed solution
Column-Family Stores
Cassandra
- Best For: Write-heavy workloads, high availability, linear scalability
- Strengths: No single point of failure, tunable consistency
- Use Cases: Time-series, IoT, messaging, activity feeds
- Trade-offs: Complex operations, eventual consistency
- When to Choose: High write volume, availability over consistency
HBase
- Best For: Big data applications, Hadoop ecosystem
- Strengths: Hadoop integration, consistent reads
- Use Cases: Analytics on big data, time-series at scale
- When to Choose: Hadoop ecosystem, very large datasets
Graph Databases
Neo4j
- Best For: Complex relationships, graph algorithms, traversals
- Strengths: Mature ecosystem, Cypher query language, algorithms
- Use Cases: Social networks, recommendation engines, fraud detection
- Trade-offs: Specialized use case, learning curve
- When to Choose: Relationship-heavy data, graph algorithms
Time-Series Databases
InfluxDB
- Best For: Time-series data, IoT, monitoring, analytics
- Strengths: Purpose-built, efficient storage, query language
- Use Cases: IoT sensors, monitoring, DevOps metrics
- When to Choose: High-volume time-series data
TimescaleDB
- Best For: Time-series with SQL familiarity
- Strengths: PostgreSQL compatibility, SQL queries, ecosystem
- Use Cases: Financial data, IoT with complex queries
- When to Choose: Time-series + SQL requirements
Search Engines
Elasticsearch
- Best For: Full-text search, log analysis, real-time search
- Strengths: Powerful search, analytics, ecosystem (ELK stack)
- Use Cases: Search applications, log analysis, monitoring
- Trade-offs: Complex operations, resource intensive
- When to Choose: Advanced search requirements, analytics
Data Warehouses
Snowflake
- Best For: Cloud-native analytics, data sharing, varied workloads
- Strengths: Separation of compute/storage, automatic scaling
- Use Cases: Data warehousing, analytics, data science
- When to Choose: Cloud-native, analytics-focused, multi-cloud
BigQuery
- Best For: Serverless analytics, Google ecosystem, machine learning
- Strengths: Serverless, petabyte scale, ML integration
- Use Cases: Analytics, data science, reporting
- When to Choose: Google Cloud, serverless analytics
Selection Criteria Matrix
| Criterion | SQL | NewSQL | Document | Key-Value | Column-Family | Graph | Time-Series |
|---|---|---|---|---|---|---|---|
| ACID Guarantees | ✅ Strong | ✅ Strong | ⚠️ Eventual | ⚠️ Eventual | ⚠️ Tunable | ⚠️ Varies | ⚠️ Varies |
| Horizontal Scaling | ❌ Limited | ✅ Native | ✅ Native | ✅ Native | ✅ Native | ⚠️ Limited | ✅ Native |
| Query Flexibility | ✅ High | ✅ High | ⚠️ Moderate | ❌ Low | ❌ Low | ✅ High | ⚠️ Specialized |
| Schema Flexibility | ❌ Rigid | ❌ Rigid | ✅ High | ✅ High | ⚠️ Moderate | ✅ High | ⚠️ Structured |
| Performance (Reads) | ⚠️ Good | ⚠️ Good | ✅ Excellent | ✅ Excellent | ✅ Excellent | ⚠️ Good | ✅ Excellent |
| Performance (Writes) | ⚠️ Good | ⚠️ Good | ✅ Excellent | ✅ Excellent | ✅ Excellent | ⚠️ Good | ✅ Excellent |
| Operational Complexity | ✅ Low | ❌ High | ⚠️ Moderate | ✅ Low | ❌ High | ⚠️ Moderate | ⚠️ Moderate |
| Ecosystem Maturity | ✅ Mature | ⚠️ Growing | ✅ Mature | ✅ Mature | ✅ Mature | ✅ Mature | ⚠️ Growing |
Decision Checklist
Requirements Analysis
- [ ] Data Volume: Current and projected data size
- [ ] Transaction Volume: Reads per second, writes per second
- [ ] Consistency Requirements: Strong vs eventual consistency needs
- [ ] Query Patterns: Simple lookups vs complex analytics
- [ ] Schema Evolution: How often does schema change?
- [ ] Geographic Distribution: Single region vs global
- [ ] Availability Requirements: Acceptable downtime
- [ ] Team Expertise: Existing knowledge and learning curve
- [ ] Budget Constraints: Licensing, infrastructure, operational costs
- [ ] Compliance Requirements: Data residency, audit trails
Technical Evaluation
- [ ] Performance Testing: Benchmark with realistic data and queries
- [ ] Scalability Testing: Test scaling limits and patterns
- [ ] Failure Scenarios: Test backup, recovery, and failure handling
- [ ] Integration Testing: APIs, connectors, ecosystem tools
- [ ] Migration Path: How to migrate from current system
- [ ] Monitoring and Observability: Available tooling and metrics
Operational Considerations
- [ ] Management Complexity: Setup, configuration, maintenance
- [ ] Backup and Recovery: Built-in vs external tools
- [ ] Security Features: Authentication, authorization, encryption
- [ ] Upgrade Path: Version compatibility and upgrade process
- [ ] Support Options: Community vs commercial support
- [ ] Lock-in Risk: Portability and vendor independence
Common Decision Patterns
E-commerce Platform
Typical Choice: PostgreSQL or MySQL
- Primary Data: Product catalog, orders, users (structured)
- Query Patterns: OLTP with some analytics
- Consistency: Strong consistency for financial data
- Scale: Moderate with read replicas
- Additional: Redis for caching, Elasticsearch for product search
IoT/Sensor Data Platform
Typical Choice: TimescaleDB or InfluxDB
- Primary Data: Time-series sensor readings
- Query Patterns: Time-based aggregations, trend analysis
- Scale: High write volume, moderate read volume
- Additional: Kafka for ingestion, PostgreSQL for metadata
Social Media Application
Typical Choice: Combination approach
- User Profiles: MongoDB (flexible schema)
- Relationships: Neo4j (graph relationships)
- Activity Feeds: Cassandra (high write volume)
- Search: Elasticsearch (content discovery)
- Caching: Redis (sessions, real-time data)
Analytics Platform
Typical Choice: Snowflake or BigQuery
- Primary Use: Complex analytical queries
- Data Volume: Large (TB to PB scale)
- Query Patterns: Ad-hoc analytics, reporting
- Users: Data analysts, data scientists
- Additional: Data lake (S3/GCS) for raw data storage
Global SaaS Application
Typical Choice: CockroachDB or DynamoDB
- Requirements: Multi-region, strong consistency
- Scale: Global user base
- Compliance: Data residency requirements
- Availability: High availability across regions
Migration Strategies
From Monolithic to Distributed
1. Assessment: Identify scaling bottlenecks 2. Data Partitioning: Plan how to split data 3. Gradual Migration: Move non-critical data first 4. Dual Writes: Run both systems temporarily 5. Validation: Verify data consistency 6. Cutover: Switch reads and writes gradually
Technology Stack Evolution
1. Start Simple: Begin with PostgreSQL or MySQL 2. Identify Bottlenecks: Monitor performance and scaling issues 3. Selective Scaling: Move specific workloads to specialized databases 4. Polyglot Persistence: Use multiple databases for different use cases 5. Service Boundaries: Align database choice with service boundaries
Conclusion
Database selection should be driven by:
1. Specific Use Case Requirements: Not all applications need the same database 2. Data Characteristics: Structure, volume, and access patterns matter 3. Non-functional Requirements: Consistency, availability, performance targets 4. Team and Organizational Factors: Expertise, operational capacity, budget 5. Evolution Path: How requirements and scale will change over time
The best database choice is often not a single technology, but a combination of databases that each excel at their specific use case within your application architecture.