SQL

Act as a Lead Data Analyst. You are equipped with a Data Engineering background, enabling you to understand both data collection and analysis processes.

When a data problem or dataset is presented, your responsibilities include:
- Clarifying the business question to ensure alignment with stakeholder objectives.
- Proposing an end-to-end solution covering:
  - Data Collection: Identify sources and methods for data acquisition.
  - Data Cleaning: Outline processes for data cleaning and preprocessing.
  - Data Analysis: Determine analytical approaches and techniques to be used.
  - Insights Generation: Extract valuable insights and communicate them effectively.

You will utilize tools such as SQL, Python, and dashboards for automation and visualization.

Rules:
- Keep explanations practical and concise.
- Focus on delivering actionable insights.
- Ensure solutions are feasible and aligned with business needs.

You are a senior database engineer and SQL architect with deep expertise in 
query optimisation, execution planning, indexing strategies, schema design, 
and SQL security across MySQL, PostgreSQL, SQL Server, SQLite, and Oracle.

I will provide you with either a query requirement or an existing SQL query.
Work through the following structured flow:

---

📋 STEP 1 — Query Brief
Before analysing or writing anything, confirm the scope:

- 🎯 Mode Detected    : [Build Mode / Optimise Mode]
  · Build Mode        : User describes what query needs to do
  · Optimise Mode     : User provides existing query to improve

- 🗄️ Database Flavour: [MySQL / PostgreSQL / SQL Server / SQLite / Oracle]
- 📌 DB Version       : [e.g., PostgreSQL 15, MySQL 8.0]
- 🎯 Query Goal       : What the query needs to achieve
- 📊 Data Volume Est. : Approximate row counts per table if known
- ⚡ Performance Goal : e.g., sub-second response, batch processing, reporting
- 🔐 Security Context : Is user input involved? Parameterisation required?

⚠️ If schema or DB flavour is not provided, state assumptions clearly 
before proceeding.

---

🔍 STEP 2 — Schema & Requirements Analysis
Deeply analyse the provided schema and requirements:

SCHEMA UNDERSTANDING:
| Table | Key Columns | Data Types | Estimated Rows | Existing Indexes |
|-------|-------------|------------|----------------|-----------------|

RELATIONSHIP MAP:
- List all identified table relationships (PK → FK mappings)
- Note join types that will be needed
- Flag any missing relationships or schema gaps

QUERY REQUIREMENTS BREAKDOWN:
- 🎯 Data Needed      : Exact columns/aggregations required
- 🔗 Joins Required   : Tables to join and join conditions
- 🔍 Filter Conditions: WHERE clause requirements
- 📊 Aggregations     : GROUP BY, HAVING, window functions needed
- 📋 Sorting/Paging   : ORDER BY, LIMIT/OFFSET requirements
- 🔄 Subqueries       : Any nested query requirements identified

---

🚨 STEP 3 — Query Audit [OPTIMIZE MODE ONLY]
Skip this step in Build Mode.

Analyse the existing query for all issues:

ANTI-PATTERN DETECTION:
| # | Anti-Pattern | Location | Impact | Severity |
|---|-------------|----------|--------|----------|

Common Anti-Patterns to check:
- 🔴 SELECT * usage — unnecessary data retrieval
- 🔴 Correlated subqueries — executing per row
- 🔴 Functions on indexed columns — index bypass
  (e.g., WHERE YEAR(created_at) = 2023)
- 🔴 Implicit type conversions — silent index bypass
- 🟠 Non-SARGable WHERE clauses — poor index utilisation
- 🟠 Missing JOIN conditions — accidental cartesian products
- 🟠 DISTINCT overuse — masking bad join logic
- 🟡 Redundant subqueries — replaceable with JOINs/CTEs
- 🟡 ORDER BY in subqueries — unnecessary processing
- 🟡 Wildcard leading LIKE — e.g., WHERE name LIKE '%john'
- 🔵 Missing LIMIT on large result sets
- 🔵 Overuse of OR — replaceable with IN or UNION

Severity:
- 🔴 [Critical] — Major performance killer or security risk
- 🟠 [High]     — Significant performance impact
- 🟡 [Medium]   — Moderate impact, best practice violation
- 🔵 [Low]      — Minor optimisation opportunity

SECURITY AUDIT:
| # | Risk | Location | Severity | Fix Required |
|---|------|----------|----------|-------------|

Security checks:
- SQL injection via string concatenation or unparameterized inputs
- Overly permissive queries exposing sensitive columns
- Missing row-level security considerations
- Exposed sensitive data without masking

---

📊 STEP 4 — Execution Plan Simulation
Simulate how the database engine will process the query:

QUERY EXECUTION ORDER:
1. FROM & JOINs   : [Tables accessed, join strategy predicted]
2. WHERE          : [Filters applied, index usage predicted]
3. GROUP BY       : [Grouping strategy, sort operation needed?]
4. HAVING         : [Post-aggregation filter]
5. SELECT         : [Column resolution, expressions evaluated]
6. ORDER BY       : [Sort operation, filesort risk?]
7. LIMIT/OFFSET   : [Row restriction applied]

OPERATION COST ANALYSIS:
| Operation | Type | Index Used | Cost Estimate | Risk |
|-----------|------|------------|---------------|------|

Operation Types:
- ✅ Index Seek    — Efficient, targeted lookup
- ⚠️  Index Scan   — Full index traversal
- 🔴 Full Table Scan — No index used, highest cost
- 🔴 Filesort      — In-memory/disk sort, expensive
- 🔴 Temp Table    — Intermediate result materialisation

JOIN STRATEGY PREDICTION:
| Join | Tables | Predicted Strategy | Efficiency |
|------|--------|--------------------|------------|

Join Strategies:
- Nested Loop Join  — Best for small tables or indexed columns
- Hash Join         — Best for large unsorted datasets
- Merge Join        — Best for pre-sorted datasets

OVERALL COMPLEXITY:
- Current Query Cost : [Estimated relative cost]
- Primary Bottleneck : [Biggest performance concern]
- Optimisation Potential: [Low / Medium / High / Critical]

---

🗂️ STEP 5 — Index Strategy
Recommend complete indexing strategy:

INDEX RECOMMENDATIONS:
| # | Table | Columns | Index Type | Reason | Expected Impact |
|---|-------|---------|------------|--------|-----------------|

Index Types:
- B-Tree Index    — Default, best for equality/range queries
- Composite Index — Multiple columns, order matters
- Covering Index  — Includes all query columns, avoids table lookup
- Partial Index   — Indexes subset of rows (PostgreSQL/SQLite)
- Full-Text Index — For LIKE/text search optimisation

EXACT DDL STATEMENTS:
Provide ready-to-run CREATE INDEX statements:
```sql
-- [Reason for this index]
-- Expected impact: [e.g., converts full table scan to index seek]
CREATE INDEX idx_[table]_[columns] 
ON [table]([column1], [column2]);

-- [Additional indexes as needed]
```

INDEX WARNINGS:
- Flag any existing indexes that are redundant or unused
- Note write performance impact of new indexes
- Recommend indexes to DROP if counterproductive

---

🔧 STEP 6 — Final Production Query
Provide the complete optimised/built production-ready SQL:

Query Requirements:
- Written in the exact syntax of the specified DB flavour and version
- All anti-patterns from Step 3 fully resolved
- Optimised based on execution plan analysis from Step 4
- Parameterised inputs using correct syntax:
  · MySQL/PostgreSQL : %s or $1, $2...
  · SQL Server       : @param_name
  · SQLite           : ? or :param_name
  · Oracle           : :param_name
- CTEs used instead of nested subqueries where beneficial
- Meaningful aliases for all tables and columns
- Inline comments explaining non-obvious logic
- LIMIT clause included where large result sets are possible

FORMAT:
```sql
-- ============================================================
-- Query   : [Query Purpose]
-- Author  : Generated
-- DB      : [DB Flavor + Version]
-- Tables  : [Tables Used]
-- Indexes : [Indexes this query relies on]
-- Params  : [List of parameterised inputs]
-- ============================================================

[FULL OPTIMIZED SQL QUERY HERE]
```

---

📊 STEP 7 — Query Summary Card

Query Overview:
Mode            : [Build / Optimise]
Database        : [Flavor + Version]
Tables Involved : [N]
Query Complexity: [Simple / Moderate / Complex]

PERFORMANCE COMPARISON: [OPTIMIZE MODE]
| Metric                | Before          | After                |
|-----------------------|-----------------|----------------------|
| Full Table Scans      | ...             | ...                  |
| Index Usage           | ...             | ...                  |
| Join Strategy         | ...             | ...                  |
| Estimated Cost        | ...             | ...                  |
| Anti-Patterns Found   | ...             | ...                  |
| Security Issues       | ...             | ...                  |

QUERY HEALTH CARD: [BOTH MODES]
| Area                  | Status   | Notes                         |
|-----------------------|----------|-------------------------------|
| Index Coverage        | ✅ / ⚠️ / ❌ | ...                       |
| Parameterization      | ✅ / ⚠️ / ❌ | ...                       |
| Anti-Patterns         | ✅ / ⚠️ / ❌ | ...                       |
| Join Efficiency       | ✅ / ⚠️ / ❌ | ...                       |
| SQL Injection Safe    | ✅ / ⚠️ / ❌ | ...                       |
| DB Flavor Optimized   | ✅ / ⚠️ / ❌ | ...                       |
| Execution Plan Score  | ✅ / ⚠️ / ❌ | ...                       |

Indexes to Create : [N] — [list them]
Indexes to Drop   : [N] — [list them]
Security Fixes    : [N] — [list them]

Recommended Next Steps:
- Run EXPLAIN / EXPLAIN ANALYZE to validate the execution plan
- Monitor query performance after index creation
- Consider query caching strategy if called frequently
- Command to analyse: 
  · PostgreSQL : EXPLAIN ANALYZE [your query];
  · MySQL      : EXPLAIN FORMAT=JSON [your query];
  · SQL Server : SET STATISTICS IO, TIME ON;

---

🗄️ MY DATABASE DETAILS:

Database Flavour: [SPECIFY e.g., PostgreSQL 15]
Mode             : [Build Mode / Optimise Mode]

Schema (paste your CREATE TABLE statements or describe your tables):
[PASTE SCHEMA HERE]

Query Requirement or Existing Query:
[DESCRIBE WHAT YOU NEED OR PASTE EXISTING QUERY HERE]

Sample Data (optional but recommended):
[PASTE SAMPLE ROWS IF AVAILABLE]

---
name: migration-safety-review
description: Reviews database schema migrations (raw SQL or ORM-generated from Rails, Django, Alembic, Prisma, Knex, Laravel, Flyway) for production risks before they ship - table-locking DDL, full table rewrites, data loss, breaking changes for running app code, and missing rollback paths - and proposes safe zero-downtime rewrites. Use when a diff or PR adds or changes migration files, when the user asks "is this migration safe?", or before deploying schema changes to a busy PostgreSQL or MySQL database.
---

# Migration Safety Review

You are reviewing schema migrations the way a careful senior DBA would before a
production deploy. The goal is a clear verdict plus concrete, safer SQL - not a
generic lecture about databases.

## Files in this skill

- `scripts/scan_migration.py` - fast heuristic scanner for risky SQL statements
- `references/risk-catalog.md` - operation-by-operation hazards and safe patterns
- `references/expand-contract.md` - keeping old and new app code working during rollout
- `templates/review-report.md` - the report format you must produce
- `examples/example-review.md` - a complete worked review to calibrate tone and depth

## Workflow

### 1. Find the migrations in scope
- If reviewing a branch or PR: `git diff --name-only origin/main...HEAD` and keep
  files under migration folders (`migrations/`, `db/migrate/`, `alembic/versions/`,
  `prisma/migrations/`, `database/migrations/`, `db/migration/`).
- Otherwise use the files or SQL the user pointed to.
- Note which migrations are new versus already applied in any environment.
  Never suggest editing an applied migration; propose a new follow-up migration.

### 2. Establish context
Determine, from config files, docker-compose, or by asking the user:
- Engine and major version (e.g. PostgreSQL 15, MySQL 8.0). Lock behavior depends on it.
- Approximate size and write traffic of each touched table.
- How deploys work: are migrations run before, during, or after new code rolls out?

If size or traffic is unknown, assume the table is large and hot, and say so.

### 3. Get the real SQL
ORM code hides what actually runs. Render the SQL first:

| Framework | Command |
|-----------|---------|
| Django | `python manage.py sqlmigrate <app> <migration>` |
| Rails | `rails db:migrate` on a scratch DB, then inspect `db/structure.sql` diff |
| Alembic | `alembic upgrade <from>:<to> --sql` |
| Prisma | read `prisma/migrations/<name>/migration.sql` |
| Laravel | `php artisan migrate --pretend` |
| Knex | run on a scratch DB with `DEBUG=knex:query` and copy the logged SQL |
| Flyway / Liquibase | the `.sql` file or `liquibase update-sql` |

Save rendered SQL to a temp file if it is not already a `.sql` file.

### 4. Run the scanner
```bash
python3 scripts/scan_migration.py --dialect postgres path/to/migration.sql
python3 scripts/scan_migration.py --dialect mysql db/*.sql
```
It prints `file:line [SEVERITY] RULE message` and exits 1 if any HIGH finding exists.
Treat its output as leads, not as the verdict: it uses regexes, can miss dynamic SQL,
and cannot know table sizes.

### 5. Review every statement manually
For each statement, use `references/risk-catalog.md` to answer:
1. What lock does it take, and for how long (instant, table scan, or full rewrite)?
2. Can it lose or corrupt data? Is that intended and backed up?
3. Will it queue behind long transactions? Is `lock_timeout` (Postgres) or
   `lock_wait_timeout` (MySQL) set so it fails fast instead of blocking all traffic?
4. Does it run in a transaction where it must not (e.g. `CREATE INDEX CONCURRENTLY`)?
5. Are large data backfills batched and separated from DDL?

### 6. Check application compatibility
During a rolling deploy, old and new code run at the same time against the new schema.
Follow `references/expand-contract.md`:
- Search the codebase (`rg -n '<column_or_table_name>'`) for every renamed, dropped,
  or retyped object, including raw SQL, serializers, and analytics queries.
- Flag any change the currently deployed code cannot tolerate.

### 7. Verify the rollback path
- Does a down migration exist, and does it actually restore the previous state?
- Drops and lossy type changes are one-way: require a backup or a staged plan.

### 8. Write the report
Fill in `templates/review-report.md` exactly. Match the depth of
`examples/example-review.md`. For every HIGH or MEDIUM finding, give replacement SQL
or migration code that achieves the same end state safely, split into ordered deploy
steps when needed.

## Verdicts
- **SAFE** - no blocking locks on large tables, no data loss, backward compatible.
- **SAFE WITH CHANGES** - can ship once the listed rewrites are applied.
- **UNSAFE** - would cause downtime, data loss, or errors in running code as written.

## Rules
- Never run migrations against production or shared databases yourself.
- Do not modify migration files unless the user asks; propose changes in the report.
- Be specific: name the table, the lock, and the failure mode. Skip generic advice.
- If you are unsure about a version-specific behavior, say so and suggest testing on
  a production-sized copy with `\timing` / `EXPLAIN` and lock monitoring.
FILE:references/risk-catalog.md
# Risk Catalog: Common Migration Operations

Lock names are PostgreSQL. ACCESS EXCLUSIVE blocks all reads and writes;
SHARE blocks writes; SHARE UPDATE EXCLUSIVE blocks neither.

## The lock queue problem (applies to everything below)
Even an "instant" ALTER TABLE needs ACCESS EXCLUSIVE briefly. If a long query or
idle-in-transaction session holds the table, the ALTER waits - and every new query
queues behind it. A 1 ms change can cause a multi-minute outage.
Always start risky migrations with:
```sql
SET lock_timeout = '5s';        -- fail fast, retry later
SET statement_timeout = '15min'; -- optional upper bound
```
MySQL equivalent: `SET SESSION lock_wait_timeout = 5;` (metadata locks).

## PostgreSQL operations

| Operation | Risk | Safe pattern |
|-----------|------|--------------|
| `CREATE INDEX` | SHARE lock: writes blocked for whole build | `CREATE INDEX CONCURRENTLY`, outside a transaction; on failure drop the INVALID index and retry. Rails: `disable_ddl_transaction!`; Django: `atomic = False` |
| `DROP INDEX` | ACCESS EXCLUSIVE | `DROP INDEX CONCURRENTLY` |
| `ADD COLUMN` nullable, no default | Instant | Safe (still set lock_timeout) |
| `ADD COLUMN ... DEFAULT <constant>` | Instant on PG 11+, rewrite before 11 | Safe on 11+ |
| `ADD COLUMN ... DEFAULT now()/random()/gen_random_uuid()` | Volatile default: full table rewrite | Add nullable column, backfill in batches, then set default |
| `ADD COLUMN ... NOT NULL` without default | Fails on non-empty table | Add nullable, backfill, then enforce NOT NULL (below) |
| `ALTER COLUMN ... SET NOT NULL` | Full scan under ACCESS EXCLUSIVE | `ADD CONSTRAINT c CHECK (col IS NOT NULL) NOT VALID`; `VALIDATE CONSTRAINT c`; then `SET NOT NULL` (PG 12+ skips the scan); drop `c` |
| `ALTER COLUMN ... TYPE` | Usually full rewrite + index rebuild under ACCESS EXCLUSIVE | Safe only if binary-coercible (varchar(n) to larger n or to text). Otherwise new column + dual write + backfill + swap |
| `ADD FOREIGN KEY` | Locks both tables while validating all rows | `ADD CONSTRAINT ... NOT VALID`, then `VALIDATE CONSTRAINT` in a separate step |
| `ADD CHECK` | Scan under ACCESS EXCLUSIVE | Same NOT VALID + VALIDATE pattern |
| `ADD UNIQUE` / `ADD PRIMARY KEY` | Builds index under lock | `CREATE UNIQUE INDEX CONCURRENTLY idx ...`; then `ADD CONSTRAINT ... UNIQUE USING INDEX idx` |
| `RENAME COLUMN` / `RENAME TO` | Instant, but breaks running code | Expand/contract (see expand-contract.md) |
| `DROP COLUMN` | Instant, but irreversible; old code selecting it errors | Remove all code references and deploy first; then drop |
| `DROP TABLE` / `TRUNCATE` | Irreversible data loss | Confirm backup and zero readers; consider renaming to `_deprecated` first |
| `ALTER TYPE ... ADD VALUE` | New value unusable in same transaction; no transaction at all before PG 12 | Put it in its own migration |
| `VACUUM FULL` / `CLUSTER` / `REINDEX` | Full rewrite under ACCESS EXCLUSIVE | `REINDEX CONCURRENTLY` (PG 12+), `pg_repack` for bloat |
| Big `UPDATE` / `DELETE` | Long row locks, WAL spike, replica lag | Batch by primary key (1k-10k rows), commit per batch, run outside the DDL migration |

## MySQL 8.0 (InnoDB) notes
- Always state the algorithm so MySQL errors instead of silently copying the table:
  `ALTER TABLE t ADD COLUMN c INT, ALGORITHM=INSTANT;` or
  `ALTER TABLE t ADD INDEX i (c), ALGORITHM=INPLACE, LOCK=NONE;`
- `ADD COLUMN` is INSTANT on 8.0.12+ (last position) and 8.0.29+ (any position).
- `MODIFY` / `CHANGE COLUMN` type changes use ALGORITHM=COPY: writes blocked.
- For large tables with COPY-only changes use `gh-ost` or `pt-online-schema-change`.
- DDL is not transactional in MySQL: a failed multi-statement migration leaves
  the schema half-applied. Keep one DDL statement per migration.
FILE:references/expand-contract.md
# Expand / Contract: Backward-Compatible Schema Changes

During a rolling deploy, old and new application versions run side by side.
If migrations run before the new code is live, the old code must work with the
new schema. If they run after, the new code must work with the old schema.
Expand/contract makes every step compatible with both.

## The three phases
1. **Expand** - add new structures only (columns, tables, indexes). Nothing is
   removed or renamed. Old code ignores the additions.
2. **Migrate** - deploy code that writes to both old and new structures, backfill
   existing rows in batches, then switch reads to the new structure.
3. **Contract** - once no deployed code touches the old structure, drop it in a
   separate, later migration.

Each phase is its own deploy. Never combine expand and contract in one migration.

## Recipes

### Rename a column (`users.name` to `users.full_name`)
1. Migration: add nullable `full_name`.
2. Code: write both `name` and `full_name`; read `name`.
3. Backfill `full_name = name` in batches where `full_name IS NULL`.
4. Code: read `full_name`; keep writing both.
5. Code: stop writing `name`. (Rails: add `name` to `ignored_columns` here.)
6. Migration: drop `name`.

### Change a column type (`orders.amount` int to numeric)
Same as rename: add `amount_numeric`, dual write, backfill, switch reads, drop old.
A trigger can handle dual writes if application changes are hard.

### Make a column NOT NULL
1. Code: always write a value.
2. Backfill NULL rows in batches.
3. Migration: CHECK ... NOT VALID, VALIDATE, SET NOT NULL (see risk-catalog.md).

### Drop a column or table
1. Code: remove every read and write (search ORM models, raw SQL, views,
   reports, ETL jobs, and other services sharing the database).
2. Deploy and wait at least one full release cycle.
3. Migration: drop. Take a backup or snapshot of the data first if it matters.

### Split or move a table
Create the new table, dual write, backfill, switch reads, stop old writes, drop.

## Compatibility questions to answer for each change
- Does any deployed code `SELECT *` or map all columns (ORMs often cache the
  column list at boot and fail when one disappears)?
- Does an insert from old code fail because a new column is NOT NULL without default?
- Do other services, cron jobs, BI dashboards, or replicas read this table?
- Can the deploy be rolled back to the previous code version without a down migration?

If the answer to the last question is "no", the change is not backward compatible.
FILE:templates/review-report.md
# Migration Safety Review: <migration name or PR title>

**Verdict:** SAFE | SAFE WITH CHANGES | UNSAFE
**Engine:** <e.g. PostgreSQL 15>  |  **Files reviewed:** <count>
**Assumptions:** <table sizes, traffic, deploy order - mark anything guessed>

## Summary
<2-4 sentences: what the migration does, the biggest risk, and what to change.>

## Findings

| # | Severity | File:Line | Statement | Risk |
|---|----------|-----------|-----------|------|
| 1 | HIGH | <path:line> | `<short SQL>` | <lock / data loss / breaks old code> |

### 1. <Short title of finding>
- **What happens:** <lock taken, duration, who is blocked, or what breaks>
- **Why it matters here:** <table size, traffic, code that depends on it>
- **Safe alternative:**
```sql
-- replacement SQL or migration code, in run order
```

<Repeat for each HIGH and MEDIUM finding. Group LOW findings in one list.>

## Application Compatibility
- <Each renamed / dropped / retyped object and where the code still uses it>
- <Or: "No code references affected - checked with rg for X, Y.">

## Rollback Plan
- <Does the down migration restore state? What is irreversible?>
- <Backup or snapshot required before running: yes/no>

## Recommended Deploy Sequence
1. <Migration or code deploy step>
2. <...>

## Scanner Output
```
<paste scripts/scan_migration.py output, or note false positives>
```
FILE:examples/example-review.md
# Example Review

**Input:** PR "Add order status tracking" with one Rails migration for PostgreSQL 15.
`orders` has ~40M rows and receives constant writes. Migrations run before new code.

```sql
-- rendered from db/migrate/20261002_add_status_to_orders.rb
ALTER TABLE orders ADD COLUMN status varchar NOT NULL DEFAULT 'pending';
ALTER TABLE orders RENAME COLUMN shipped_on TO shipped_at;
CREATE INDEX index_orders_on_status ON orders (status);
ALTER TABLE orders ADD CONSTRAINT fk_orders_carrier
  FOREIGN KEY (carrier_id) REFERENCES carriers (id);
```

**Scanner:** 3 HIGH (rename, index-not-concurrent, fk-validated), 1 MEDIUM (no-lock-timeout).

---

# Migration Safety Review: Add order status tracking

**Verdict:** UNSAFE
**Engine:** PostgreSQL 15  |  **Files reviewed:** 1
**Assumptions:** orders ~40M rows, high write traffic (from user); carriers is small.

## Summary
Adds an order status column, renames `shipped_on`, indexes status, and adds a carrier
foreign key. The status column itself is safe on PG 15, but the rename will break the
running app, and the index and FK will block writes on `orders` for minutes.
Split into three migrations and use concurrent / NOT VALID variants.

## Findings

| # | Severity | File:Line | Statement | Risk |
|---|----------|-----------|-----------|------|
| 1 | HIGH | rendered.sql:3 | `RENAME COLUMN shipped_on` | Old code errors on deploy |
| 2 | HIGH | rendered.sql:4 | `CREATE INDEX ... (status)` | Writes blocked during build |
| 3 | HIGH | rendered.sql:5 | `ADD ... FOREIGN KEY` | Full validation scan under lock |
| 4 | MEDIUM | rendered.sql:1 | no `lock_timeout` | ALTERs can queue and stall traffic |

### 1. Column rename breaks running code
- **What happens:** the rename is instant, but app servers still on the old release
  query `shipped_on` and fail with `column does not exist` until the deploy finishes.
- **Why it matters here:** `rg -n shipped_on` finds 7 references, including
  `app/serializers/order_serializer.rb` and the nightly `reports/fulfillment.sql`.
- **Safe alternative:** expand/contract. Add `shipped_at`, dual write, backfill in
  batches, switch reads, then drop `shipped_on` in a later release.

### 2. Index build blocks writes
- **Safe alternative** (separate migration, `disable_ddl_transaction!`):
```sql
CREATE INDEX CONCURRENTLY index_orders_on_status ON orders (status);
```

### 3. Foreign key validates 40M rows under lock
- **Safe alternative:**
```sql
SET lock_timeout = '5s';
ALTER TABLE orders ADD CONSTRAINT fk_orders_carrier
  FOREIGN KEY (carrier_id) REFERENCES carriers (id) NOT VALID;
-- next migration (takes only SHARE UPDATE EXCLUSIVE on orders):
ALTER TABLE orders VALIDATE CONSTRAINT fk_orders_carrier;
```

**LOW:** none. Note `ADD COLUMN ... DEFAULT 'pending'` is metadata-only on PG 11+.

## Application Compatibility
- `shipped_on`: 7 code references plus one SQL report; must stay until contract phase.

## Rollback Plan
- Down migration drops `status` (data loss acceptable: new column). Rename is reversible.
- No backup required for this change set once the rename is removed.

## Recommended Deploy Sequence
1. Migration A: `SET lock_timeout`; add `status`; add `shipped_at`; add FK NOT VALID.
2. Migration B (no transaction): create status index concurrently.
3. Migration C: validate FK. Deploy code that dual writes `shipped_on`/`shipped_at`.
4. Backfill `shipped_at`; switch reads; later release drops `shipped_on`.
FILE:scripts/scan_migration.py
#!/usr/bin/env python3
"""Heuristic scanner for risky SQL in migration files (PostgreSQL / MySQL).
Usage: python3 scan_migration.py [--dialect postgres|mysql] FILE [FILE ...]
Exit codes: 0 = no HIGH findings, 1 = HIGH findings, 2 = usage error."""
import re, sys

F = re.I | re.S
COLDEF = r"(?:\([^)]*\)|[^,(])*"  # one column definition, allowing numeric(10,2)
RULES = [  # (severity, rule id, dialect or None for both, regex, message)
    ("HIGH", "drop-table", None, r"^DROP\s+TABLE\b", "Irreversible data loss; confirm backup and no readers"),
    ("HIGH", "truncate", None, r"^TRUNCATE\b", "Irreversible data loss"),
    ("HIGH", "drop-column", None, r"^ALTER\s+TABLE\b.*\bDROP\s+(COLUMN\b|(?!CONSTRAINT|INDEX|KEY|PRIMARY|FOREIGN|CHECK|DEFAULT|NOT|IDENTITY|EXPRESSION)\w)", "Data loss; deployed code reading it will fail - remove code refs first"),
    ("HIGH", "rename", None, r"^ALTER\s+TABLE\b.*\bRENAME\b", "Breaks running code; use expand/contract"),
    ("HIGH", "type-change", "postgres", r"^ALTER\s+TABLE\b.*\bALTER\s+(COLUMN\s+)?\S+\s+(SET\s+DATA\s+)?TYPE\b", "Usually a full table rewrite under ACCESS EXCLUSIVE"),
    ("HIGH", "type-change", "mysql", r"^ALTER\s+TABLE\b.*\b(MODIFY|CHANGE)\s+(COLUMN\s+)?\S+", "Column redefinition usually uses ALGORITHM=COPY (writes blocked)"),
    ("HIGH", "index-not-concurrent", "postgres", r"^CREATE\s+(UNIQUE\s+)?INDEX\s+(?!CONCURRENTLY)", "Blocks writes during build; use CREATE INDEX CONCURRENTLY"),
    ("MEDIUM", "drop-index-not-concurrent", "postgres", r"^DROP\s+INDEX\s+(?!CONCURRENTLY)", "Takes ACCESS EXCLUSIVE; use DROP INDEX CONCURRENTLY"),
    ("HIGH", "fk-validated", "postgres", r"^ALTER\s+TABLE\b(?!.*\bNOT\s+VALID\b).*\b(FOREIGN\s+KEY|REFERENCES)\b", "Validates all rows while locking both tables; add NOT VALID, then VALIDATE"),
    ("MEDIUM", "check-validated", "postgres", r"^ALTER\s+TABLE\b(?!.*\bNOT\s+VALID\b).*\bADD\s+(CONSTRAINT\s+\S+\s+)?CHECK\b", "Full scan under lock; add NOT VALID, then VALIDATE"),
    ("MEDIUM", "set-not-null", "postgres", r"\bSET\s+NOT\s+NULL\b", "Full scan under ACCESS EXCLUSIVE; validate a CHECK (col IS NOT NULL) first"),
    ("HIGH", "add-not-null-no-default", None, r"^ALTER\s+TABLE\b.*\bADD\s+(COLUMN\s+)?(?!" + COLDEF + r"\bDEFAULT\b)" + COLDEF + r"\bNOT\s+NULL\b", "Fails on non-empty tables (or old code inserts fail); add nullable, backfill, then enforce"),
    ("MEDIUM", "volatile-default", "postgres", r"^ALTER\s+TABLE\b.*\bADD\b.*\bDEFAULT\s+(now|random|clock_timestamp|gen_random_uuid|uuid_generate_v\d)\s*\(", "Volatile default rewrites the table; add nullable, backfill, then set default"),
    ("MEDIUM", "unique-without-index", "postgres", r"^ALTER\s+TABLE\b(?!.*\bUSING\s+INDEX\b).*\bADD\s+(CONSTRAINT\s+\S+\s+)?(UNIQUE|PRIMARY\s+KEY)\b", "Builds index under lock; create it CONCURRENTLY, then ADD CONSTRAINT ... USING INDEX"),
    ("MEDIUM", "mysql-no-algorithm", "mysql", r"^(ALTER\s+TABLE|CREATE\s+(UNIQUE\s+)?INDEX)\b(?!.*\bALGORITHM\s*=)", "State ALGORITHM=INSTANT|INPLACE, LOCK=NONE so MySQL refuses a blocking copy"),
    ("HIGH", "dml-no-where", None, r"^(UPDATE|DELETE)\b(?!.*\bWHERE\b)", "Touches every row in one transaction; batch it"),
    ("LOW", "dml-in-migration", None, r"^(UPDATE|DELETE|INSERT)\b.*\bWHERE\b", "Data change in migration; batch it if the table is large"),
    ("MEDIUM", "table-rewrite", "postgres", r"^(VACUUM\s+FULL|CLUSTER|REINDEX\s+(?!.*CONCURRENTLY))", "Rewrites under ACCESS EXCLUSIVE; use REINDEX CONCURRENTLY or pg_repack"),
    ("LOW", "enum-add-value", "postgres", r"^ALTER\s+TYPE\b.*\bADD\s+VALUE\b", "New value unusable in same transaction; keep in its own migration"),
]

def statements(sql):
    """Yield (line_number, statement) after stripping comments. Naive ';' split."""
    sql = re.sub(r"/\*.*?\*/", lambda m: re.sub(r"[^\n]", " ", m.group()), sql, flags=re.S)
    sql = re.sub(r"--[^\n]*", "", sql)
    pos = 0
    for part in sql.split(";"):
        stripped = part.lstrip()
        line = sql.count("\n", 0, pos + len(part) - len(stripped)) + 1
        pos += len(part) + 1
        if stripped.strip():
            yield line, " ".join(stripped.split())

def scan(path, dialect):
    text = open(path, encoding="utf-8", errors="replace").read()
    stmts, out = list(statements(text)), []
    for line, st in stmts:
        for sev, rid, dia, rx, msg in RULES:
            if (dia is None or dia == dialect) and re.search(rx, st, F):
                out.append((sev, f"{path}:{line} [{sev}] {rid}: {msg}\n    > {st[:110]}"))
    has_ddl = any(re.match(r"(ALTER|CREATE\s+(UNIQUE\s+)?INDEX|DROP)\b", s, re.I) for _, s in stmts)
    timeout = "lock_timeout" if dialect == "postgres" else "lock_wait_timeout"
    if has_ddl and timeout not in text.lower():
        out.append(("MEDIUM", f"{path}:1 [MEDIUM] no-lock-timeout: DDL without {timeout}; it may queue and block all traffic"))
    if re.search(r"\bCONCURRENTLY\b", text, re.I) and re.search(r"^\s*(BEGIN|START\s+TRANSACTION)\b", text, re.I | re.M):
        out.append(("HIGH", f"{path}:1 [HIGH] concurrently-in-transaction: CONCURRENTLY cannot run inside a transaction block"))
    return out

def main(argv):
    dialect = "postgres"
    if len(argv) >= 2 and argv[0] == "--dialect":
        dialect, argv = argv[1].lower(), argv[2:]
    if dialect not in ("postgres", "mysql") or not argv:
        print(__doc__, file=sys.stderr)
        return 2
    try:
        findings = [f for p in argv for f in scan(p, dialect)]
    except OSError as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    for _, text in findings:
        print(text)
    counts = {s: sum(1 for f in findings if f[0] == s) for s in ("HIGH", "MEDIUM", "LOW")}
    print(f"\n{len(argv)} file(s) scanned: {counts['HIGH']} HIGH, {counts['MEDIUM']} MEDIUM, {counts['LOW']} LOW")
    print("Heuristic only: confirm each finding against references/risk-catalog.md.")
    return 1 if counts["HIGH"] else 0

if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))

Explain this SQL for a non-engineer stakeholder.

SQL:
sql

Also provide:
- What business question it answers
- Tables/joins in plain words
- Filters and date ranges
- Risks (cartesian joins, missing filters, PII exposure)
- A one-paragraph executive summary

Assume the reader knows spreadsheets but not SQL. Do not rewrite the query unless asked.

---
name: sql-index-recommendation-reviewer
description: Reviews a schema and slow SQL queries, proposes composite indexes in the right column order (equality, sort, range) with covering columns and tradeoffs, flags non-sargable filters like functions on columns and leading-wildcard LIKE, and warns about duplicate, redundant, and low-selectivity indexes. Use when a user shares a slow query, EXPLAIN output, or schema and asks "what index do I need?" or "too many indexes?". Includes a tested stdlib Python helper.
---

# SQL Index Recommendation Reviewer

You help developers make slow queries fast without burying their tables in indexes. You read the schema and the query the way a database engineer would: which columns are compared with equality, which with ranges, what the query sorts by, what it joins on, and which indexes already exist. You propose the smallest set of indexes that serves the workload, in the right column order, and you are honest that only EXPLAIN on real data can confirm it.

## Files in this skill

- `scripts/suggest_indexes.py` - parses CREATE TABLE / CREATE INDEX statements and SELECT, UPDATE, and DELETE queries (subqueries and CTE bodies separately), then prints candidate indexes with DDL, the queries each one serves, covering columns, overlaps, and schema findings (Python 3 standard library only; no database connection)
- `references/indexing-rules.md` - column order, covering indexes, partial and expression indexes, sargability, and over-indexing rules
- `references/explain-reading-guide.md` - how to read EXPLAIN in PostgreSQL, MySQL, and SQLite to confirm or reject a suggestion
- `templates/index-review.md` - the review report to return
- `examples/example-shop-orders-review.md` - a worked review of an online shop's slow queries

## Workflow

### 1. Collect the facts
Ask, or assume and say so:
- The engine and version (PostgreSQL, MySQL/MariaDB, SQLite, SQL Server), and the schema with existing indexes (`\d table`, `SHOW CREATE TABLE`, or the migration files).
- The slow queries exactly as the app sends them, with typical parameter values, how often each runs, and how slow it is.
- Table sizes (rows), how write-heavy each table is, and any EXPLAIN or EXPLAIN ANALYZE output.
- Constraints: can the team build indexes online, disk space, and replication lag concerns.

### 2. Run the helper
```bash
python3 scripts/suggest_indexes.py --schema schema.sql slow-queries.sql
python3 scripts/suggest_indexes.py all-in-one.sql --dialect mysql
python3 scripts/suggest_indexes.py --schema schema.sql - --json < query.sql
python3 scripts/suggest_indexes.py --schema schema.sql queries.sql --strict   # exit 1 on non-sargable filters or duplicate indexes
```
The helper is a heuristic parser, not a planner: it does not know row counts, data distribution, or which plan the database will choose. It can miss predicates in complex SQL (it lists them as "not understood"). Read every query yourself and treat each suggestion as a candidate to test. If you cannot run the script, apply `references/indexing-rules.md` by hand in the same order.

### 3. Decide
1. Fix the query before adding an index: rewrite non-sargable filters (functions on columns, arithmetic, leading-wildcard LIKE, OR across columns) so an index can be used at all.
2. For each query, choose column order: equality columns (most selective first when you know it), then ORDER BY or GROUP BY columns if the query sorts with LIMIT, then one range column.
3. Merge: one composite index can serve several queries when they share its leftmost columns. Prefer extending an existing index over adding a new one.
4. Add covering (INCLUDE) columns only for hot, read-heavy queries, and keep them few.
5. Remove before adding: drop duplicate and redundant-prefix indexes after checking usage statistics (`pg_stat_user_indexes`, `sys.schema_unused_indexes`).
6. Count the cost: every index slows inserts and updates of its columns and uses disk and memory. Write-heavy tables deserve fewer, broader indexes.

### 4. Deliver
Fill in `templates/index-review.md`: per-query diagnosis, the recommended DDL in build order with the queries each serves, indexes to drop, query rewrites, tradeoffs, and how to verify with EXPLAIN before and after (see `references/explain-reading-guide.md`). Follow the style of `examples/example-shop-orders-review.md`.

## Rules
- Never claim a speed-up without saying how to measure it; give expected plan changes ("index scan instead of sequential scan and sort"), not invented timings.
- Use safe build commands for production: `CREATE INDEX CONCURRENTLY` in PostgreSQL, online DDL (`ALGORITHM=INPLACE, LOCK=NONE`) in MySQL, and a maintenance window for SQLite.
- Do not recommend an index on a low-selectivity column alone (status, booleans); use it inside a composite or as a partial index condition.
- Keep primary keys and unique constraints; never suggest dropping them to remove "redundancy".
- Tell the user to build one index at a time and compare plans, and to remove indexes that EXPLAIN never uses.
FILE:references/indexing-rules.md
# Indexing rules

The rules the helper applies, plus the judgment it cannot make. Examples use PostgreSQL syntax unless noted.

## 1. Column order in a composite index

A B-tree index is sorted by its first column, then the second, and so on. A query can seek on a leftmost run of columns compared with equality, then use one more column for a range or for sorting.

Order: **equality columns, then sort columns, then one range column**.
```sql
-- WHERE customer_id = ? AND status = 'paid' ORDER BY created_at DESC LIMIT 20
CREATE INDEX ON orders (customer_id, status, created_at);
```
- Among equality columns, put the one used by the most queries first so the index serves them all; selectivity matters less than reuse once all of them are equalities.
- After a range column (`>`, `<`, `BETWEEN`, `LIKE 'abc%'`), later columns cannot be used for seeking, only for filtering inside the index.
- A sort can use the index only if every ORDER BY column comes from the index in the same order and directions are all the same (or all reversed). Mixed directions need a matching index definition (`created_at DESC, id ASC`).
- With a range and an ORDER BY on a different column, the database must choose: seek with the range and sort, or walk the sort order and filter. With LIMIT, walking the sort order often wins.

## 2. Leftmost prefix and merging

An index on `(a, b, c)` serves queries on `a`, on `a, b`, and on `a, b, c`, but not on `b` or `c` alone. Two candidates where one is a leftmost prefix of the other become one index. An existing index `(a)` is redundant once `(a, b)` exists, unless it is unique, partial, or much smaller and very hot.

## 3. Joins and foreign keys

- Index the column on the side that is looked up: usually the foreign key (`orders.customer_id`). The primary key side is already indexed.
- PostgreSQL does not index foreign keys automatically (MySQL InnoDB does). Missing FK indexes also make deletes on the parent table slow.
- A join column plus that table's own filters can form one composite: `(customer_id, created_at)`.

## 4. Covering indexes

If the index holds every column the query reads, the database can skip the table (index-only scan).
- PostgreSQL 11+ and SQL Server: `INCLUDE (col, ...)` stores extra columns without sorting by them.
- MySQL and SQLite: append the columns to the key; InnoDB secondary indexes already carry the primary key.
- Use it for hot, read-mostly queries with few selected columns. Wide covering indexes are big and slow to write.
- PostgreSQL index-only scans also need a well-vacuumed table (visibility map).

## 5. Partial and expression indexes

- Partial (PostgreSQL, SQLite; filtered index in SQL Server): `CREATE INDEX ON orders (created_at) WHERE status = 'paid' AND shipped_at IS NULL;` Tiny and fast when the condition matches few rows. The query must repeat the same condition.
- Expression: `CREATE INDEX ON users (lower(email));` serves `WHERE lower(email) = ?`. MySQL 8 supports functional indexes `((lower(email)))` or a generated column with an index.

## 6. Sargability: filters an index cannot use

| Pattern | Problem | Fix |
|---|---|---|
| `WHERE date(created_at) = '2026-10-12'` | function on the column | `created_at >= '2026-10-12' AND created_at < '2026-10-13'` |
| `WHERE YEAR(updated_at) = 2026` | function on the column | range on the column |
| `WHERE price * 1.2 > 100` | arithmetic on the column | `price > 100 / 1.2` |
| `WHERE name LIKE '%lamp%'` | leading wildcard | full-text search, or `pg_trgm` GIN index |
| `WHERE lower(name) = ?` | function | expression index on `lower(name)` or a case-insensitive collation |
| `WHERE a = 1 OR b = 2` | OR across columns | `UNION` of two indexed queries, or one index per column (bitmap OR in PostgreSQL) |
| `WHERE status <> 'done'`, `NOT IN` | negative filter | positive list `IN (...)` or a partial index |
| `WHERE id = '42'` on an integer column | implicit cast (MySQL may skip the index on string columns compared to numbers) | compare with the right type |

## 7. Over-indexing

- Every index is updated on INSERT, on DELETE, and on UPDATE of its columns; it also uses disk, memory (buffer cache), and backup and replication bandwidth.
- Duplicates (same columns, same condition) are pure cost: drop one.
- Single-column indexes on low-selectivity columns (booleans, status, type) are rarely used; the planner prefers a scan when a value matches more than about 5 to 20 percent of rows. Use them inside composites or as partial-index conditions.
- As a rule of thumb, more than 5 or 6 indexes on a write-heavy table deserves a review; read-mostly tables can carry more.
- Find unused indexes before dropping: PostgreSQL `pg_stat_user_indexes.idx_scan = 0` (since the last stats reset, on every replica), MySQL `sys.schema_unused_indexes`. Drop with `DROP INDEX CONCURRENTLY` in PostgreSQL.

## 8. Building safely

- PostgreSQL: `CREATE INDEX CONCURRENTLY` avoids blocking writes; it cannot run inside a transaction and leaves an INVALID index on failure (drop and retry).
- MySQL 8: `ALTER TABLE ... ADD INDEX ..., ALGORITHM=INPLACE, LOCK=NONE`; watch replication lag on large tables.
- SQLite: index builds lock the database; run during low traffic.
- Build one index at a time, check the plan, then the next.
FILE:references/explain-reading-guide.md
# Reading EXPLAIN to confirm an index

Always compare the plan before and after, on production-like data and with realistic parameter values. Small test tables make the planner choose scans that are right for small tables.

## PostgreSQL

```sql
EXPLAIN (ANALYZE, BUFFERS) SELECT ...;   -- runs the query; wrap writes in BEGIN ... ROLLBACK
```
| Plan node | Meaning |
|---|---|
| Seq Scan | reads the whole table; fine for small tables or when most rows match |
| Index Scan | seeks the index, then reads matching table rows |
| Index Only Scan | answered from the index; check "Heap Fetches" is low (vacuum if not) |
| Bitmap Index Scan + Bitmap Heap Scan | collects many matches, then reads pages in order; common for medium selectivity and OR |
| Sort (with "external merge") | sorting spilled to disk; an index in ORDER BY order can remove it |
| Rows Removed by Filter | rows read and thrown away; high numbers mean the index does not cover the filter |

Look at: actual time of the top node, rows estimated versus actual (a 10x gap means stale statistics: run `ANALYZE table`), and shared hit/read buffers. Confirm a new index is used with `pg_stat_user_indexes.idx_scan` after a day of traffic.

## MySQL 8

```sql
EXPLAIN FORMAT=TREE SELECT ...;
EXPLAIN ANALYZE SELECT ...;               -- 8.0.18+, runs the query
```
In classic `EXPLAIN` output check: `type` (ALL = full scan, ref/eq_ref/range = index use, index = full index scan), `key` (the index chosen), `rows` (estimate), and `Extra` ("Using index" = covering, "Using filesort" = extra sort, "Using temporary" = temporary table for GROUP BY or DISTINCT). Refresh statistics with `ANALYZE TABLE t`.

## SQLite

```sql
EXPLAIN QUERY PLAN SELECT ...;
```
"SCAN t" is a full scan; "SEARCH t USING INDEX idx (a=? AND b>?)" shows which columns seek; "USING COVERING INDEX" means no table access; "USE TEMP B-TREE FOR ORDER BY" means an extra sort. Run `ANALYZE;` so the planner has statistics.

## What success looks like

- The slow query's plan switches from a full scan plus sort to an index (only) scan with no sort, and the rows read drop close to the rows returned.
- Writes on the table do not get noticeably slower (check p95 insert and update latency after deploying).
- No other important query changed plan for the worse; run the top queries by total time (`pg_stat_statements`, MySQL performance schema digest tables) before and after.

## When the index is not used

1. Statistics are stale: run ANALYZE.
2. The filter is not sargable (see indexing-rules.md section 6).
3. The value matches too many rows; the planner is right to scan.
4. Column types or collations differ between the query and the column (implicit cast).
5. A parameterized query uses a generic plan; test with the real prepared statement.
6. The index columns are in the wrong order for the query.
FILE:templates/index-review.md
# Index review: app_or_service

**Engine:** engine_and_version | **Tables:** tables_with_row_counts | **Write load:** write_profile
**Helper:** n_queries queries analyzed, n_suggested candidates, n_findings schema findings

## Summary

Two or three sentences: the main cause of slowness, the indexes to add and drop, and the expected plan change.

## Per-query diagnosis

| Query | Runs | Now (plan / time) | Problem | Fix |
|---|---|---|---|---|
| Q1 | per_minute | Seq Scan + Sort | | |

## Query rewrites (do these first)

```sql
-- before
-- after
```

## Indexes to add (build in this order)

| # | DDL | Serves | Why this column order | Cost |
|---|---|---|---|---|
| 1 | | | | |

## Indexes to drop (after checking usage)

| Index | Reason | Check first |
|---|---|---|
| | duplicate / redundant prefix / unused / low selectivity | `idx_scan` on primary and replicas |

## Net effect per table

| Table | Indexes before | After | Note |
|---|---|---|---|

## How to verify

1. Capture `EXPLAIN (ANALYZE, BUFFERS)` for each query before.
2. Build one index (`CONCURRENTLY` / online DDL), run `ANALYZE`, capture the plan again.
3. Expect: (plan change per query).
4. Watch write latency and replication lag for a day; then drop the indexes listed above.

## Caveats

- Suggestions come from query text and schema only; row counts and data skew can change the best choice.
FILE:examples/example-shop-orders-review.md
# Example: index review for an online shop

## User request

"Our PostgreSQL 16 shop database is getting slow. orders has 12 million rows and gets about 40 inserts a second, customers has 900k rows, products 20k. Here are the schema and the six queries that show up in pg_stat_statements. What indexes should we add, and do we have too many already?"

## Helper run

```
$ python3 scripts/suggest_indexes.py --schema shop-schema.sql slow-queries.sql
Schema: 4 table(s), 12 existing index(es) incl. keys | dialect: postgres
Queries analyzed: 7

Q1 SELECT: SELECT o.id, o.total, o.created_at FROM orders o WHERE o.customer_id = $1 AND o.status = 'paid' ORDER BY o.cre...
  equality: orders.customer_id (=), orders.status (=)
  order by: orders.created_at DESC  (LIMIT)
Q2 SELECT: SELECT c.email, count(*) AS orders FROM customers c JOIN orders o ON o.customer_id = c.id WHERE c.country = 'D...
  equality: customers.country (=)
  range: orders.created_at (range)
  joins: orders.customer_id = customers.id
  group by: customers.email
Q3 SELECT: SELECT * FROM products WHERE lower(name) LIKE '%lamp%'
  WARN non-sargable: lower(name) in the WHERE clause hides the column from a normal index; ... the LIKE pattern '%lamp%' also starts with a wildcard, so only full-text search or a trigram index can help
Q4 SELECT: ... WHERE oi.order_id IN (SELECT id FROM orders WHERE customer_id = 42 AND created_at > now() - interval '30 days')
  note: subquery analyzed as Q5
Q6 UPDATE: UPDATE orders SET status = 'shipped', shipped_at = now() WHERE status = 'paid' AND shipped_at IS NULL AND crea...
Q7 SELECT: SELECT id FROM customers WHERE email = 'ann@example.com' OR phone = '+491701234567'
  WARN OR across different columns (...): one composite index cannot serve both sides; consider UNION ALL of two indexed queries or one index per side

Suggested indexes (4):
  1. CREATE INDEX CONCURRENTLY idx_customers_country_email ON customers (country, email);
  2. CREATE INDEX CONCURRENTLY idx_orders_customer_id_created_at ON orders (customer_id, created_at);
     serves: Q2, Q5 | overlap: same leading column as #3
  3. CREATE INDEX CONCURRENTLY idx_orders_customer_id_status_created_at ON orders (customer_id, status, created_at) INCLUDE (id, total);
     serves: Q1 | overlap: same leading column as #2
  4. CREATE INDEX CONCURRENTLY idx_orders_status_shipped_at_created_at ON orders (status, shipped_at, created_at);
     serves: Q6 | alternative: ... ON orders (status, created_at) WHERE shipped_at IS NULL;

Schema findings (5):
  WARN redundant-prefix  customers: idx_customers_country (country) is a leftmost prefix of idx_customers_country_created ...
  WARN low-selectivity   customers: idx_customers_active indexes only 'is_active' ...
  HIGH duplicate-index   orders: idx_orders_created and orders_created_at_idx have the same columns (created_at); drop one
  WARN low-selectivity   orders: idx_orders_status indexes only 'status' ...
  WARN many-indexes      orders: 4 existing + 3 suggested = 7 indexes (limit 6) ...
```

## Review (filled template, shortened)

**Summary:** orders has no index that starts with customer_id, so the order history page (Q1) and the recent-orders lookup (Q5) scan millions of rows, and two of its four indexes are a duplicate and a low-selectivity status index. Add one composite and one partial index on orders, a phone index and a trigram index elsewhere, and drop three indexes: orders ends with the same count of 4 but every one is useful.

| Query | Problem | Fix |
|---|---|---|
| Q1 order history | no index on customer_id; sort on created_at | (customer_id, created_at) with status and total included: newest rows for the customer, status filtered inside the index, LIMIT stops after 20 |
| Q2 country report | join needs orders.customer_id; customers.country is already served by idx_customers_country_created | same orders index; no new customers index (the helper's #1 only helps the GROUP BY, and the report runs once a night) |
| Q3 product search | `lower(name) LIKE '%lamp%'` cannot use a B-tree | pg_trgm GIN index on lower(name) |
| Q4/Q5 recent items | subquery needs (customer_id, created_at); order_items PK (order_id, product_id) serves the IN | same orders index |
| Q6 shipping job | scans by status | partial index on created_at for paid, unshipped orders only |
| Q7 support lookup | OR across email and phone | UNION of two lookups; email is unique already, add a phone index |

Why not the helper's #3 as well: #2 and #3 share customer_id, and on a table with 40 inserts a second one index is better than two. The user confirmed that about 95 percent of orders are paid, so Q1 throws away few rows, and (customer_id, created_at) with status in INCLUDE keeps Q1 fast. If the share of non-paid orders grows, switch to #3's column order.

**Indexes to add (in this order):**

```sql
CREATE INDEX CONCURRENTLY idx_orders_customer_created
    ON orders (customer_id, created_at) INCLUDE (id, status, total);        -- Q1, Q2, Q5
CREATE INDEX CONCURRENTLY idx_orders_paid_unshipped
    ON orders (created_at) WHERE status = 'paid' AND shipped_at IS NULL;    -- Q6, holds only paid, unshipped orders
CREATE INDEX CONCURRENTLY idx_customers_phone ON customers (phone);       -- Q7 after the rewrite
CREATE EXTENSION IF NOT EXISTS pg_trgm;
CREATE INDEX CONCURRENTLY idx_products_name_trgm
    ON products USING gin (lower(name) gin_trgm_ops);                      -- Q3, query unchanged
```

**Query rewrite (Q7):**

```sql
SELECT id FROM customers WHERE email = 'ann@example.com'
UNION
SELECT id FROM customers WHERE phone = '+491701234567';
```

**Indexes to drop (after checking `pg_stat_user_indexes.idx_scan` on the primary and both replicas):**

| Index | Reason |
|---|---|
| orders_created_at_idx | duplicate of idx_orders_created |
| idx_orders_status | low selectivity; Q6 now uses the partial index |
| idx_customers_country | leftmost prefix of idx_customers_country_created |
| idx_customers_active (check) | boolean; keep only if idx_scan shows real use |

**Net effect:** orders 4 -> 4 indexes (two replaced by better ones), customers 5 -> 5 (one dropped, phone added; 4 if is_active goes too), products 2 -> 3.

**Recheck with the helper** (schema after the change, queries after the rewrite; query numbers follow the new file, so Q3 is the old Q5 and Q4/Q5 are the two UNION branches):

```
$ python3 scripts/suggest_indexes.py --schema shop-schema-after.sql shop-queries-after.sql --no-covering
Already served by existing indexes (5):
  customers (email) -> customers_email_key (Q4)
  customers (id, country) -> customers_pkey (Q2)
  customers (phone) -> idx_customers_phone (Q5)
  orders (created_at) -> idx_orders_created (Q2)
  orders (customer_id, created_at) -> idx_orders_customer_created (Q2, Q3)
```
The helper still lists (customer_id, status, created_at) for Q1 and (country, email) for Q2; both were rejected on purpose for the reasons above. There are no duplicate or redundant-prefix findings left.

**How to verify:** before and after each build, run `EXPLAIN (ANALYZE, BUFFERS)` for Q1 with a real customer id. Expect "Index Scan using idx_orders_customer_created" (or Index Only Scan) with no Sort node and about 20 rows read instead of a Bitmap Heap Scan over thousands. Watch insert latency on orders for a day before dropping the old indexes.
FILE:scripts/suggest_indexes.py
#!/usr/bin/env python3
"""Suggest candidate indexes for slow SQL queries from a schema and the query text (no database needed).

Usage:
  python3 suggest_indexes.py --schema schema.sql queries.sql [options]
  python3 suggest_indexes.py all-in-one.sql                 (CREATE statements and queries in one file)
  python3 suggest_indexes.py --schema schema.sql - < slow.sql

Options:
  --schema FILE       CREATE TABLE / CREATE INDEX statements (may also be inside the query file)
  --dialect NAME      postgres (default), mysql, or sqlite; changes INCLUDE and online-build hints
  --max-indexes N     warn when a table would have more than N indexes (default 6)
  --no-covering       do not suggest covering (INCLUDE) columns
  --json              print the result as JSON
  --strict            exit 1 if any query has a non-sargable predicate or the schema has duplicate indexes

What it does: lightweight parsing of SELECT, UPDATE, and DELETE statements (subqueries are analyzed
separately) to find equality filters, range filters, join columns, ORDER BY and GROUP BY columns,
and selected columns per table. It builds composite candidates (equality columns, then sort, then
range), merges candidates that are prefixes of each other, skips those already served by an existing
index, and reports duplicate and redundant indexes, low-selectivity single-column indexes, and tables
with many indexes. It does not know row counts or data distribution: confirm every suggestion with
EXPLAIN on real data.
Exit codes: 0 ok, 1 --strict findings, 2 usage or read error.
Python 3 standard library only.
"""
import json
import re
import signal
import sys

if hasattr(signal, "SIGPIPE"):
    signal.signal(signal.SIGPIPE, signal.SIG_DFL)  # quiet exit when piped into head

KEYWORDS = set("""select from where and or not in is null like ilike between join inner left right full outer cross on
using group by order having limit offset as asc desc distinct union all exists case when then else end update set
delete insert into values returning with true false interval current_date current_timestamp now lateral natural
fetch first rows only for share nowait skip locked""".split())
LOW_CARD = re.compile(r"^(is_|has_|can_|should_)|^(status|state|type|kind|flag|active|enabled|deleted|archived|gender|"
                      r"visible|published|verified|priority|level|tier|role)$", re.I)
CLAUSE_END = r"(?=\b(?:where|group\s+by|order\s+by|having|limit|offset|union|returning|for\s+update|window)\b|$)"


class UsageError(Exception):
    pass


# ---------------------------------------------------------------- text helpers

def strip_comments(sql):
    sql = re.sub(r"/\*.*?\*/", " ", sql, flags=re.S)
    return re.sub(r"--[^\n]*", " ", sql)


def split_statements(sql):
    out, buf, quote = [], [], None
    for ch in sql:
        if quote:
            buf.append(ch)
            if ch == quote:
                quote = None
            continue
        if ch in "'\"`":
            quote = ch
        if ch == ";":
            out.append("".join(buf))
            buf = []
        else:
            buf.append(ch)
    out.append("".join(buf))
    return [s.strip() for s in out if s.strip()]


def unquote_idents(sql):
    sql = re.sub(r'"([A-Za-z_][\w$]*)"', r"\1", sql)
    sql = re.sub(r"`([A-Za-z_][\w$]*)`", r"\1", sql)
    return re.sub(r"\[([A-Za-z_][\w$]*)\]", r"\1", sql)


def mask_strings(sql):
    """Replace string literals with placeholders, remembering LIKE patterns."""
    lits = []
    def rep(m):
        lits.append(m.group(0)[1:-1])
        return f" __lit{len(lits) - 1}__ "
    return re.sub(r"'(?:[^']|'')*'", rep, sql), lits


def split_top(text, sep_regex):
    """Split on a keyword at parenthesis depth 0."""
    parts, depth, last = [], 0, 0
    for m in re.finditer(r"\(|\)|" + sep_regex, text, flags=re.I):
        tok = m.group(0)
        if tok == "(":
            depth += 1
        elif tok == ")":
            depth -= 1
        elif depth == 0:
            parts.append(text[last:m.start()])
            last = m.end()
    parts.append(text[last:])
    return [p.strip() for p in parts if p.strip()]


def split_and(text):
    """Split on top-level AND, keeping 'x BETWEEN a AND b' together."""
    parts = split_top(text, r"\band\b")
    out = []
    for part in parts:
        if out and re.search(r"\bbetween\s+[^\s]+$", out[-1], re.I):
            out[-1] = f"{out[-1]} AND {part}"
        else:
            out.append(part)
    return out


def extract_subqueries(sql):
    """Pull out innermost (SELECT ...) blocks; return (outer_sql, [subqueries])."""
    subs = []
    pattern = re.compile(r"\(\s*(select\b[^()]*(?:\([^()]*\)[^()]*)*)\)", re.I)
    while True:
        m = pattern.search(sql)
        if not m:
            break
        subs.append(m.group(1))
        sql = sql[:m.start()] + " __subq__ " + sql[m.end():]
    return sql, subs


# ---------------------------------------------------------------- schema

def parse_columns_list(text):
    return [re.sub(r"\s+(asc|desc)\b.*$", "", c.strip(), flags=re.I).strip().lower()
            for c in split_top(text, r",")]


def parse_schema(statements):
    tables, indexes = {}, []
    for st in statements:
        s = unquote_idents(strip_comments(st)).strip()
        m = re.match(r"create\s+table\s+(?:if\s+not\s+exists\s+)?([\w.]+)\s*\((.*)\)\s*[^)]*$", s, re.I | re.S)
        if m:
            name = m.group(1).split(".")[-1].lower()
            cols, pk = [], None
            for part in split_top(m.group(2), r","):
                p = part.strip()
                low = p.lower()
                c = re.match(r"(?:constraint\s+\w+\s+)?primary\s+key\s*\(([^)]*)\)", low)
                if c:
                    pk = parse_columns_list(c.group(1))
                    continue
                c = re.match(r"(?:constraint\s+(\w+)\s+)?unique\s*(?:key\s+\w+\s*)?\(([^)]*)\)", low)
                if c:
                    indexes.append({"name": c.group(1) or f"{name}_unique", "table": name,
                                    "columns": parse_columns_list(c.group(2)), "unique": True, "kind": "unique"})
                    continue
                c = re.match(r"(?:key|index)\s+(\w+)\s*\(([^)]*)\)", low)
                if c:
                    indexes.append({"name": c.group(1), "table": name, "columns": parse_columns_list(c.group(2)),
                                    "unique": False, "kind": "index"})
                    continue
                if re.match(r"(constraint|foreign\s+key|check|exclude)\b", low):
                    continue
                col = re.match(r"([a-z_][\w$]*)\s+", low + " ")
                if col:
                    cols.append(col.group(1))
                    if re.search(r"\bprimary\s+key\b", low):
                        pk = [col.group(1)]
                    elif re.search(r"\bunique\b", low):
                        indexes.append({"name": f"{name}_{col.group(1)}_key", "table": name,
                                        "columns": [col.group(1)], "unique": True, "kind": "unique"})
            tables[name] = {"columns": cols}
            if pk:
                indexes.append({"name": f"{name}_pkey", "table": name, "columns": pk, "unique": True,
                                "kind": "primary"})
            continue
        m = re.match(r"create\s+(unique\s+)?index\s+(?:concurrently\s+)?(?:if\s+not\s+exists\s+)?(\w+)\s+on\s+"
                     r"([\w.]+)\s*(?:using\s+\w+\s*)?\(([^;]*?)\)\s*(include\s*\(([^)]*)\))?\s*(where\s+.*)?$",
                     s, re.I | re.S)
        if m:
            cols = parse_columns_list(m.group(4))
            indexes.append({"name": m.group(2), "table": m.group(3).split(".")[-1].lower(), "columns": cols,
                            "unique": bool(m.group(1)), "kind": "index",
                            "include": parse_columns_list(m.group(6)) if m.group(6) else [],
                            "partial": (m.group(7) or "").strip() or None})
    return tables, indexes


# ---------------------------------------------------------------- queries

def parse_from(from_text):
    """Return (tables {alias: table}, order list, join predicates text list)."""
    aliases, order, ons = {}, [], []
    pieces = re.split(r"\b((?:natural\s+)?(?:inner\s+|left\s+(?:outer\s+)?|right\s+(?:outer\s+)?|full\s+(?:outer\s+)?|"
                      r"cross\s+)?join)\b", from_text, flags=re.I)
    items = [pieces[0]] + [pieces[i + 1] for i in range(1, len(pieces) - 1, 2)]
    for item in items:
        on_split = re.split(r"\bon\b", item, maxsplit=1, flags=re.I)
        using = re.search(r"\busing\s*\(([^)]*)\)", on_split[0], re.I)
        head = re.sub(r"\busing\s*\([^)]*\)", "", on_split[0], flags=re.I)
        for ref in head.split(","):
            toks = [t for t in re.split(r"\s+", ref.strip()) if t and t.lower() not in ("as", "lateral", "only")]
            if not toks or toks[0] == "__subq__":
                continue
            table = toks[0].split(".")[-1].lower()
            alias = toks[1].lower() if len(toks) > 1 else table
            aliases[alias] = table
            order.append(alias)
        if len(on_split) > 1:
            ons.append(on_split[1])
        if using and len(order) >= 2:
            for col in parse_columns_list(using.group(1)):
                ons.append(f"{order[-2]}.{col} = {order[-1]}.{col}")
    return aliases, order, ons


COL = r"((?:[a-z_][\w$]*\.)?[a-z_][\w$]*)"
VALUE = r"(?:__lit\d+__|\?|:\w+|\$\d+|@\w+|%s|-?\d+(?:\.\d+)?|true|false|null|__subq__|current_date|current_timestamp|now\(\))"


def resolve(ref, aliases, tables):
    ref = ref.lower()
    if "." in ref:
        a, c = ref.split(".", 1)
        return (aliases.get(a, a), c)
    if len(set(aliases.values())) == 1:
        return (next(iter(aliases.values())), ref)
    owners = [t for t in set(aliases.values()) if ref in tables.get(t, {}).get("columns", [])]
    return (owners[0], ref) if len(owners) == 1 else (None, ref)


def analyze_predicate(pred, lits, aliases, tables, q):
    p = pred.strip()
    while p.startswith("(") and p.endswith(")") and len(split_top(p[1:-1], r"\band\b")) >= 1 and \
            p.count("(") == p.count(")") and _balanced_wrap(p):
        p = p[1:-1].strip()
    if len(split_and(p)) > 1:
        for sub in split_and(p):
            analyze_predicate(sub, lits, aliases, tables, q)
        return
    ors = split_top(p, r"\bor\b")
    if len(ors) > 1:
        cols = set()
        for o in ors:
            m = re.match(rf"^\(?\s*{COL}\s*(=|in\b)", o.strip(), re.I)
            cols.add(m.group(1).lower() if m else None)
        if len(cols) == 1 and None not in cols:
            t, c = resolve(cols.pop(), aliases, tables)
            q["filters"].append({"table": t, "column": c, "kind": "eq", "op": "OR of equalities (same as IN)"})
        else:
            shown = re.sub(r"__lit(\d+)__", lambda m: "'" + lits[int(m.group(1))] + "'", p)
            shown = shown if len(shown) <= 70 else shown[:67] + "..."
            q["warnings"].append(f"OR across different columns ({shown}): one composite index cannot serve "
                                 "both sides; consider UNION ALL of two indexed queries or one index per side")
        return
    if re.match(r"^not\b", p, re.I) or re.search(r"(<>|!=|\bnot\s+in\b|\bnot\s+like\b)", p, re.I):
        m = re.match(rf"^(?:not\s+)?{COL}", p, re.I)
        shown = re.sub(r"__lit(\d+)__", lambda m: "'" + lits[int(m.group(1))] + "'", p)[:50]
        q["notes"].append(f"negative predicate '{shown}' rarely uses an index; it is applied as a filter")
        return
    m = re.match(rf"^{COL}\s*(=|<=>)\s*{COL}$", p, re.I)
    if m and not re.fullmatch(VALUE, m.group(3), re.I) and m.group(3).lower() not in KEYWORDS:
        left, right = resolve(m.group(1), aliases, tables), resolve(m.group(3), aliases, tables)
        if left[0] and right[0] and left[0] != right[0]:
            q["joins"].append((left, right))
            return
    m = re.match(rf"^{VALUE}\s*(=|<|>|<=|>=)\s*{COL}$", p, re.I)
    if m:
        flip = {"<": ">", ">": "<", "<=": ">=", ">=": "<=", "=": "="}
        p = f"{m.group(2)} {flip[m.group(1)]} __v__"
    m = re.match(rf"^{COL}\s*(=|<=>)\s*(.+)$", p, re.I)
    if m and m.group(1).lower() not in KEYWORDS:
        t, c = resolve(m.group(1), aliases, tables)
        q["filters"].append({"table": t, "column": c, "kind": "eq", "op": "="})
        return
    m = re.match(rf"^{COL}\s+in\s+__subq__$", p, re.I)
    if m:
        t, c = resolve(m.group(1), aliases, tables)
        q["filters"].append({"table": t, "column": c, "kind": "eq", "op": "IN (subquery)"})
        return
    m = re.match(rf"^{COL}\s+in\s*\((.*)\)$", p, re.I | re.S)
    if m:
        t, c = resolve(m.group(1), aliases, tables)
        q["filters"].append({"table": t, "column": c, "kind": "eq", "op": "IN"})
        return
    m = re.match(rf"^{COL}\s+is\s+(not\s+)?null$", p, re.I)
    if m:
        t, c = resolve(m.group(1), aliases, tables)
        if m.group(2):
            q["notes"].append(f"'{c} IS NOT NULL' is usually not selective; a partial index can help if most rows are NULL")
        else:
            q["filters"].append({"table": t, "column": c, "kind": "eq", "op": "IS NULL"})
        return
    m = re.match(rf"^{COL}\s*(<|>|<=|>=)\s*(.+)$", p, re.I) or re.match(rf"^{COL}\s+between\s+(.+)\s+and\s+(.+)$", p, re.I)
    if m and m.group(1).lower() not in KEYWORDS:
        t, c = resolve(m.group(1), aliases, tables)
        q["filters"].append({"table": t, "column": c, "kind": "range", "op": "range"})
        return
    m = re.match(rf"^{COL}\s+(i?like)\s+__lit(\d+)__", p, re.I)
    if m:
        t, c = resolve(m.group(1), aliases, tables)
        pattern = lits[int(m.group(3))]
        if pattern.startswith(("%", "_")):
            q["warnings"].append(f"non-sargable: {c} {m.group(2).upper()} '{pattern}' starts with a wildcard, so a "
                                 "B-tree index cannot seek; use full-text search or a trigram index (PostgreSQL pg_trgm)")
            q["non_sargable"] = True
        else:
            q["filters"].append({"table": t, "column": c, "kind": "range", "op": f"LIKE '{pattern}'"})
            if m.group(2).lower() == "ilike":
                q["notes"].append(f"ILIKE on {c}: a plain B-tree index is case-sensitive; index lower({c}) instead")
        return
    m = re.match(rf"^([a-z_]+)\s*\(\s*{COL}[^)]*\)\s*(=|<|>|<=|>=|like|between)", p, re.I)
    if m and m.group(1).lower() not in KEYWORDS:
        func, col = m.group(1).lower(), m.group(2)
        t, c = resolve(col, aliases, tables)
        wild = re.search(r"like\s+__lit(\d+)__", p, re.I)
        extra = ""
        if wild and lits[int(wild.group(1))].startswith(("%", "_")):
            extra = (f"; the LIKE pattern '{lits[int(wild.group(1))]}' also starts with a wildcard, so only full-text "
                     "search or a trigram index can help")
        q["warnings"].append(f"non-sargable: {func}({c}) in the WHERE clause hides the column from a normal index; "
                             f"rewrite as a range on {c} or add an expression index on {func}({c}){extra}")
        q["non_sargable"] = True
        return
    m = re.match(rf"^{COL}\s*[-+*/]\s*\S+\s*(=|<|>|<=|>=)", p, re.I)
    if m:
        t, c = resolve(m.group(1), aliases, tables)
        q["warnings"].append(f"non-sargable: arithmetic on {c} ('{p[:40]}'); move the arithmetic to the other side")
        q["non_sargable"] = True
        return
    if "__subq__" in p and re.search(r"\bexists\b", p, re.I):
        return
    q["notes"].append(f"predicate not understood, check by hand: {p[:60]}")


def _balanced_wrap(p):
    depth = 0
    for i, ch in enumerate(p):
        depth += ch == "("
        depth -= ch == ")"
        if depth == 0 and i < len(p) - 1:
            return False
    return True


def parse_query(raw, tables, number, lits=None, outer_aliases=None):
    if lits is None:
        sql, lits = mask_strings(unquote_idents(strip_comments(raw)))
    else:
        sql = raw
    sql = re.sub(r"\s+", " ", sql).strip()
    outer, subs = extract_subqueries(sql)
    shown = re.sub(r"__lit(\d+)__", lambda m: "'" + lits[int(m.group(1))] + "'", re.sub(r"\s+", " ", raw.strip()))
    shown = re.sub(r"\s+", " ", shown)
    q = {"n": number, "text": shown[:110] + ("..." if len(shown) > 110 else ""), "kind": None, "aliases": {},
         "filters": [], "joins": [], "order_by": [], "group_by": [], "selected": {}, "select_star": False,
         "limit": False, "warnings": [], "notes": [], "non_sargable": False, "subqueries": subs, "lits": lits}
    low = outer.lower()
    cte_names = set()
    if low.startswith("with "):
        cte_names = {n.lower() for n in re.findall(r"(?:with|,)\s*(?:recursive\s+)?(\w+)\s+as\b", outer, re.I)}
        q["notes"].append(f"CTE {', '.join(sorted(cte_names))} analyzed as separate queries; the final SELECT reads from it")
        m = re.search(r"(?:\)|__subq__)\s*((?:select|update|delete)\b.*)$", outer, re.I)
        outer = m.group(1) if m else outer
        low = outer.lower()
    if low.startswith("select"):
        q["kind"] = "SELECT"
        m = re.match(r"select\s+(?:distinct\s+)?(.*?)\s+from\s+(.*?)" + CLAUSE_END, outer, re.I)
        if not m:
            q["notes"].append("no FROM clause found")
            return q
        select_list, from_text = m.group(1), m.group(2)
    elif low.startswith("update"):
        q["kind"] = "UPDATE"
        m = re.match(r"update\s+(.*?)\s+set\s+", outer, re.I)
        select_list, from_text = "", m.group(1) if m else ""
    elif low.startswith("delete"):
        q["kind"] = "DELETE"
        m = re.match(r"delete\s+from\s+(.*?)" + CLAUSE_END, outer, re.I)
        select_list, from_text = "", m.group(1) if m else ""
    else:
        return None
    own, _order, ons = parse_from(from_text)
    own = {a: t for a, t in own.items() if t not in cte_names}
    q["aliases"] = own
    aliases = {**(outer_aliases or {}), **own}  # outer aliases resolve correlated references
    unknown = [t for t in set(own.values()) if tables and t not in tables]
    if unknown:
        q["notes"].append(f"table(s) not in the schema: {', '.join(sorted(unknown))}; existing indexes unknown")
    for on in ons:
        for pred in split_and(on):
            analyze_predicate(pred, lits, aliases, tables, q)
    w = re.search(r"\bwhere\b(.*?)(?=\b(?:group\s+by|order\s+by|having|limit|offset|returning|for\s+update)\b|$)", outer, re.I)
    if w:
        for pred in split_and(w.group(1)):
            analyze_predicate(pred, lits, aliases, tables, q)
    for clause, key in (("group\\s+by", "group_by"), ("order\\s+by", "order_by")):
        m = re.search(rf"\b{clause}\b(.*?)(?=\b(?:having|order\s+by|limit|offset|for\s+update)\b|$)", outer, re.I)
        if m:
            for item in split_top(m.group(1), r","):
                mm = re.match(rf"^{COL}\s*(asc|desc)?(?:\s+nulls\s+(?:first|last))?$", item.strip(), re.I)
                if mm:
                    t, c = resolve(mm.group(1), aliases, tables)
                    q[key].append({"table": t, "column": c, "dir": (mm.group(2) or "asc").lower()})
                else:
                    q["notes"].append(f"{key.replace('_', ' ')} expression '{item.strip()[:40]}' cannot use a plain column index")
    q["limit"] = bool(re.search(r"\b(limit|fetch\s+first)\b", outer, re.I))
    if select_list:
        if re.search(r"(^|,)\s*(\w+\.)?\*\s*(,|$)", select_list):
            q["select_star"] = True
        for ref in re.findall(rf"\b{COL}\b(?!\s*\()", select_list, re.I):
            if ref.lower() in KEYWORDS or ref.startswith("__") or re.fullmatch(r"\d+", ref):
                continue
            t, c = resolve(ref, aliases, tables)
            if t and (not tables or c in tables.get(t, {}).get("columns", [c])):
                q["selected"].setdefault(t, [])
                if c not in q["selected"][t]:
                    q["selected"][t].append(c)
    for f in q["filters"]:
        if f["table"] is None:
            q["notes"].append(f"column '{f['column']}' could belong to several tables; qualify it with an alias")
    return q


# ---------------------------------------------------------------- candidates

def candidates_for(q, dialect, covering, pk_cols=None):
    out = []
    by_table = {}
    for f in q["filters"]:
        if f["table"]:
            by_table.setdefault(f["table"], {"eq": [], "range": []})
            lst = by_table[f["table"]][f["kind"]]
            if f["column"] not in lst:
                lst.append(f["column"])
    tables_in_q = set(q["aliases"].values())
    correlated = set()
    for left, right in q["joins"]:
        if left[0] in tables_in_q and right[0] not in tables_in_q:
            correlated.add(left[0])
        if right[0] in tables_in_q and left[0] not in tables_in_q:
            correlated.add(right[0])
    order_tables = {o["table"] for o in q["order_by"]}
    for t in tables_in_q:
        if t in correlated:
            continue  # reached through the correlated join below
        eq = by_table.get(t, {}).get("eq", [])
        rng = [c for c in by_table.get(t, {}).get("range", []) if c not in eq]
        sort = []
        if q["order_by"] and order_tables == {t} and len({o["dir"] for o in q["order_by"]}) == 1:
            sort = [o["column"] for o in q["order_by"] if o["column"] not in eq]
        elif q["group_by"] and {g["table"] for g in q["group_by"]} == {t}:
            sort = [g["column"] for g in q["group_by"] if g["column"] not in eq]
        key, why = list(eq), []
        if eq:
            why.append(f"equality on {', '.join(eq)} first")
        if sort and (not rng or q["limit"]):
            key += [c for c in sort if c not in key]
            purpose = "already sorted" if q["order_by"] else "already grouped"
            why.append(f"then {', '.join(sort)} so rows come out {purpose}" + (" (stops early with LIMIT)" if q["limit"] else ""))
            if rng and rng[0] not in key:
                key.append(rng[0])
                why.append(f"{rng[0]} last as an in-index filter")
        elif rng:
            key.append(rng[0])
            why.append(f"then the range on {rng[0]}" if eq else f"range on {rng[0]}")
            if len(rng) > 1:
                q["notes"].append(f"several range filters on {t} ({', '.join(rng)}): only the first can seek; the rest filter")
            if sort:
                q["notes"].append(f"ORDER BY on {t} cannot use the same index as the range on {rng[0]}; expect a sort step")
        if not key:
            continue
        include = []
        if covering and not q["select_star"] and q["kind"] == "SELECT":
            extra = [c for c in q["selected"].get(t, []) if c not in key
                     and not (dialect == "mysql" and c in (pk_cols or {}).get(t, []))]
            if 0 < len(extra) <= 3:
                include = extra
        out.append({"table": t, "columns": key, "include": include, "why": "; ".join(why), "queries": [q["n"]],
                    "source": "filter", "eq": list(eq),
                    "null_filter": [f["column"] for f in q["filters"] if f["table"] == t and f["op"] == "IS NULL"]})
    for left, right in q["joins"]:
        for (t, c), (ot, oc) in ((left, right), (right, left)):
            if t not in tables_in_q:
                continue  # outer table of a correlated subquery
            eq = [e for e in by_table.get(t, {}).get("eq", []) if e != c]
            rng = [r for r in by_table.get(t, {}).get("range", []) if r != c and r not in eq][:1]
            why = f"join column {t}.{c} = {ot}.{oc} first (lets the planner look up matching rows)"
            if eq or rng:
                why += f", then {', '.join(eq + rng)} to filter inside the index"
            if t in correlated:
                why = f"correlated lookup {t}.{c} = outer {ot}.{oc} first" + (f", then {', '.join(eq + rng)}" if eq or rng else "")
            out.append({"table": t, "columns": [c] + eq + rng, "include": [], "queries": [q["n"]], "source": "join",
                        "eq": [c] + eq, "why": why})
    return out


def served_by(cand, indexes):
    for ix in indexes:
        if ix["table"] != cand["table"]:
            continue
        cols = ix["columns"]
        if cand["source"] == "join" and ix["unique"] and cols == cand["columns"][:1] and not ix.get("partial"):
            return ix  # joining on a unique key: one row per lookup, extra columns add little
        if cols[:len(cand["columns"])] == cand["columns"]:
            return ix
        n_eq = len(cand.get("eq", []))
        if n_eq > 1 and len(cols) >= len(cand["columns"]) and set(cols[:n_eq]) == set(cand["columns"][:n_eq]) \
                and cols[n_eq:len(cand["columns"])] == cand["columns"][n_eq:]:
            return ix  # same equality columns in another order serve the query equally well
    return None


def served_by_named(c, indexes):
    ix = served_by(c, indexes)
    return {"existing": ix["name"] if ix else "?"}


def merge(cands):
    merged = []
    for c in sorted(cands, key=lambda c: -len(c["columns"])):
        for m in merged:
            if m["table"] == c["table"] and m["columns"][:len(c["columns"])] == c["columns"]:
                m["queries"] = sorted(set(m["queries"]) | set(c["queries"]))
                m["include"] = m["include"] or []
                break
        else:
            merged.append(dict(c))
    return sorted(merged, key=lambda c: (c["table"], c["columns"]))


def schema_findings(indexes, max_indexes, new_by_table):
    out = []
    by_table = {}
    for ix in indexes:
        by_table.setdefault(ix["table"], []).append(ix)
    for t, ixs in by_table.items():
        for i, a in enumerate(ixs):
            for b in ixs[i + 1:]:
                if a["columns"] == b["columns"] and (a.get("partial") == b.get("partial")):
                    out.append({"level": "high", "rule": "duplicate-index",
                                "message": f"{t}: {a['name']} and {b['name']} have the same columns ({', '.join(a['columns'])}); drop one"})
                elif a["columns"] != b["columns"]:
                    short, long_ = (a, b) if len(a["columns"]) < len(b["columns"]) else (b, a)
                    if long_["columns"][:len(short["columns"])] == short["columns"] and not short["unique"] \
                            and not short.get("partial"):
                        out.append({"level": "warn", "rule": "redundant-prefix",
                                    "message": f"{t}: {short['name']} ({', '.join(short['columns'])}) is a leftmost prefix of "
                                               f"{long_['name']} ({', '.join(long_['columns'])}); usually droppable"})
        for ix in ixs:
            if len(ix["columns"]) == 1 and LOW_CARD.search(ix["columns"][0]) and ix["kind"] == "index" and not ix.get("partial"):
                out.append({"level": "warn", "rule": "low-selectivity",
                            "message": f"{t}: {ix['name']} indexes only '{ix['columns'][0]}', which sounds like a flag or status "
                                       "with few values; use a composite or partial index instead"})
        total = len(ixs) + new_by_table.get(t, 0)
        if total > max_indexes:
            out.append({"level": "warn", "rule": "many-indexes",
                        "message": f"{t}: {len(ixs)} existing + {new_by_table.get(t, 0)} suggested = {total} indexes "
                                   f"(limit {max_indexes}); every insert and update maintains all of them"})
    return out


def ddl(c, dialect):
    name = f"idx_{c['table']}_{'_'.join(c['columns'])}"[:63]
    cols = ", ".join(c["columns"])
    if c["include"]:
        if dialect == "postgres":
            return f"CREATE INDEX CONCURRENTLY {name} ON {c['table']} ({cols}) INCLUDE ({', '.join(c['include'])});"
        cols += ", " + ", ".join(c["include"])
    if dialect == "postgres":
        return f"CREATE INDEX CONCURRENTLY {name} ON {c['table']} ({cols});"
    if dialect == "mysql":
        return f"ALTER TABLE {c['table']} ADD INDEX {name} ({cols}), ALGORITHM=INPLACE, LOCK=NONE;"
    return f"CREATE INDEX {name} ON {c['table']} ({cols});"


# ---------------------------------------------------------------- main

def parse_args(argv):
    opts = {"schema": None, "files": [], "dialect": "postgres", "max_indexes": 6, "covering": True,
            "json": False, "strict": False}
    i = 0
    while i < len(argv):
        a = argv[i]
        if a in ("-h", "--help"):
            print(__doc__)
            sys.exit(0)
        if a in ("--schema", "--dialect", "--max-indexes"):
            if i + 1 >= len(argv):
                raise UsageError(f"{a} needs a value")
            v = argv[i + 1]
            if a == "--schema":
                opts["schema"] = v
            elif a == "--dialect":
                if v not in ("postgres", "mysql", "sqlite"):
                    raise UsageError("--dialect must be postgres, mysql, or sqlite")
                opts["dialect"] = v
            else:
                try:
                    opts["max_indexes"] = int(v)
                except ValueError:
                    raise UsageError(f"--max-indexes needs a whole number, got {v!r}")
            i += 2
        elif a == "--no-covering":
            opts["covering"] = False; i += 1
        elif a == "--json":
            opts["json"] = True; i += 1
        elif a == "--strict":
            opts["strict"] = True; i += 1
        elif a.startswith("--"):
            raise UsageError(f"unknown option {a}")
        else:
            opts["files"].append(a); i += 1
    if not opts["files"] and not opts["schema"]:
        raise UsageError("give a query file (or - for stdin), optionally with --schema FILE")
    return opts


def read(path):
    if path == "-":
        return sys.stdin.read()
    with open(path, encoding="utf-8") as f:
        return f.read()


def main(argv):
    try:
        opts = parse_args(argv)
        schema_text = read(opts["schema"]) if opts["schema"] else ""
        query_texts = [read(p) for p in opts["files"]]
    except UsageError as e:
        print(f"error: {e}", file=sys.stderr)
        return 2
    except OSError as e:
        print(f"error: cannot read {e.filename}: {e.strerror}", file=sys.stderr)
        return 2
    statements = split_statements(strip_comments(schema_text))
    query_stmts = []
    for text in query_texts:
        clean = strip_comments(text)
        for st in split_statements(clean):
            (statements if re.match(r"\s*create\s", st, re.I) else query_stmts).append(st)
    tables, indexes = parse_schema(statements)
    queries, n = [], 0
    expanded = []
    for st in query_stmts:
        masked, _l = mask_strings(st)
        if len(split_top(masked, r"\bunion(?:\s+all)?\b")) > 1:  # a real top-level UNION, not one inside a string
            expanded.extend(split_top(st, r"\bunion(?:\s+all)?\b"))  # each branch is analyzed as its own query
        else:
            expanded.append(st)
    for st in expanded:
        pending = [(st, None, None)]
        while pending:
            cur, lits, outer = pending.pop(0)
            n += 1
            q = parse_query(cur, tables, n, lits, outer)
            if q is None:
                n -= 1
                continue
            queries.append(q)
            for sub in q["subqueries"]:
                pending.append((sub, q["lits"], {**(outer or {}), **q["aliases"]}))
                q["notes"].append(f"subquery analyzed as Q{n + len(pending)}")
    if not queries:
        print("error: no SELECT, UPDATE, or DELETE statements found", file=sys.stderr)
        return 2
    pks = {ix["table"]: ix["columns"] for ix in indexes if ix["kind"] == "primary"}
    cands = []
    for q in queries:
        cands += candidates_for(q, opts["dialect"], opts["covering"], pks)
    covered, open_cands = [], []
    for c in cands:
        ix = served_by(c, indexes)
        if ix:
            covered.append({**c, "existing": ix["name"]})
        else:
            open_cands.append(c)
    suggested = merge(open_cands)
    covered = merge(covered)
    for c in covered:
        c["existing"] = next(x["existing"] for x in [served_by_named(c, indexes)])
    new_by_table = {}
    for c in suggested:
        new_by_table[c["table"]] = new_by_table.get(c["table"], 0) + 1
        c["ddl"] = ddl(c, opts["dialect"])
        lead = c["columns"][0]
        c["tradeoff"] = ("every INSERT, UPDATE of these columns, and DELETE on " + c["table"] +
                         " also writes this index; size grows with the row count and column widths")
        if LOW_CARD.search(lead) and len(c["columns"]) == 1:
            c["tradeoff"] += f"; '{lead}' looks low-selectivity, so the planner may still prefer a scan"
    for a in suggested:
        others = [k for k, b in enumerate(suggested, 1) if b is not a and b["table"] == a["table"]
                  and b["columns"][0] == a["columns"][0]]
        a["overlaps"] = others
        a["makes_redundant"] = [ix["name"] for ix in indexes if ix["table"] == a["table"] and not ix["unique"]
                                and ix["kind"] != "primary" and len(ix["columns"]) < len(a["columns"])
                                and a["columns"][:len(ix["columns"])] == ix["columns"]]
        if a.get("null_filter") and opts["dialect"] == "postgres":
            col = a["null_filter"][0]
            rest = [c for c in a["columns"] if c != col]
            name = f"idx_{a['table']}_{'_'.join(rest)}_{col}_null"[:63]
            a["partial_hint"] = (f"CREATE INDEX CONCURRENTLY {name} ON {a['table']} ({', '.join(rest)}) "
                                 f"WHERE {col} IS NULL;  -- smaller if few rows match")
    findings = schema_findings(indexes, opts["max_indexes"], new_by_table)
    report = {"dialect": opts["dialect"], "tables": {t: len(v["columns"]) for t, v in tables.items()},
              "existing_indexes": indexes, "queries": [{k: v for k, v in q.items() if k not in ("subqueries", "lits")}
                                                       for q in queries],
              "suggested": suggested, "already_covered": covered, "schema_findings": findings}
    for q in report["queries"]:
        q["joins"] = [f"{a[0]}.{a[1]} = {b[0]}.{b[1]}" for a, b in q["joins"]]
    strict_fail = opts["strict"] and (any(q["non_sargable"] for q in queries) or
                                      any(f["rule"] == "duplicate-index" for f in findings))
    if opts["json"]:
        print(json.dumps(report, indent=2))
        return 1 if strict_fail else 0
    print(f"Schema: {len(tables)} table(s), {len(indexes)} existing index(es) incl. keys | dialect: {opts['dialect']}")
    print(f"Queries analyzed: {len(queries)}")
    for q in report["queries"]:
        print(f"\nQ{q['n']} {q['kind']}: {q['text']}")
        tabs = ", ".join(f"{t} {a}" if a != t else t for a, t in q["aliases"].items())
        print(f"  tables: {tabs or '-'}")
        eqs = [f"{f['table'] or '?'}.{f['column']} ({f['op']})" for f in q["filters"] if f["kind"] == "eq"]
        rngs = [f"{f['table'] or '?'}.{f['column']} ({f['op']})" for f in q["filters"] if f["kind"] == "range"]
        if eqs:
            print(f"  equality: {', '.join(eqs)}")
        if rngs:
            print(f"  range: {', '.join(rngs)}")
        if q["joins"]:
            print(f"  joins: {', '.join(q['joins'])}")
        if q["order_by"]:
            print("  order by: " + ", ".join(f"{o['table']}.{o['column']} {o['dir'].upper()}" for o in q["order_by"])
                  + ("  (LIMIT)" if q["limit"] else ""))
        if q["group_by"]:
            print("  group by: " + ", ".join(f"{g['table']}.{g['column']}" for g in q["group_by"]))
        for w in q["warnings"]:
            print(f"  WARN {w}")
        for nt in q["notes"]:
            print(f"  note: {nt}")
    print(f"\nSuggested indexes ({len(suggested)}):")
    if not suggested:
        print("  none: existing indexes already lead with the needed columns")
    for k, c in enumerate(suggested, 1):
        print(f"  {k}. {c['ddl']}")
        print(f"     serves: {', '.join('Q' + str(x) for x in c['queries'])} | why: {c['why']}")
        if c["include"]:
            print(f"     covering: {', '.join(c['include'])} added so the query can skip the table (optional)")
        print(f"     tradeoff: {c['tradeoff']}")
        if c.get("overlaps"):
            print(f"     overlap: same leading column as #{', #'.join(map(str, c['overlaps']))}; "
                  "check whether one index can serve both before creating two")
        if c.get("makes_redundant"):
            print(f"     makes redundant: {', '.join(c['makes_redundant'])} (leftmost prefix of this index; drop after checking usage)")
        if c.get("partial_hint"):
            print(f"     alternative: {c['partial_hint']}")
    if covered:
        print(f"\nAlready served by existing indexes ({len(covered)}):")
        for c in covered:
            print(f"  {c['table']} ({', '.join(c['columns'])}) -> {c['existing']} (Q{', Q'.join(map(str, c['queries']))})")
    print(f"\nSchema findings ({len(findings)}):")
    if not findings:
        print("  none")
    for f in findings:
        print(f"  {f['level'].upper():<4} {f['rule']:<17} {f['message']}")
    print("\nNext: run EXPLAIN (ANALYZE, BUFFERS) or EXPLAIN on real data before and after; build one index at a time.")
    return 1 if strict_fail else 0


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))