마이그레이션 안전성 검토 프롬프트

--- name: migration-safety-review description: 데이터베이스 스키마 마이그레이션(원시 SQL 또는 Rails, Django, Alembic, Prisma, Knex, Laravel, Flyway에서 생성된 ORM)을 프로덕션 배포 전에 검토합니다 - 테이블 잠금 DDL, 전체 테이블 재작성, 데이터 손실, 실행 중인 앱 코드에 대한 중단 변경, 및 누락된 롤백 경로 - 그리고 안전한 제로 다운타임 재작성을 제안합니다. diff 또는 PR이 마이그레이션 파일을 추가하거나 변경할 때, 사용자가 "이 마이그레이션이 안전한가요?"라고 물을 때, 또는 바쁜 PostgreSQL 또는 MySQL 데이터베이스에 스키마 변경을 배포하기 전에 사용하세요. --- # 마이그레이션 안전성 검토 당신은 프로덕션 배포 전에 신중한 시니어 DBA가 하는 것처럼 스키마 마이그레이션을 검토합니다. 목표는 명확한 판정과 구체적이고 더 안전한 SQL입니다 - 데이터베이스에 대한 일반적인 강의가 아닙니다. ## 이 스킬의 파일들 - `scripts/scan_migration.py` - 위험한 SQL 문을 위한 빠른 휴리스틱 스캐너 - `references/risk-catalog.md` - 작업별 위험 요소와 안전한 패턴 - `references/expand-contract.md` - 롤아웃 중에 기존 및 새 앱 코드를 계속 작동시키기 - `templates/review-report.md` - 당신이 만들어야 하는 보고서 형식 - `examples/example-review.md` - 톤과 깊이를 보정하기 위한 완전한 작업 검토 예시 ## 워크플로우 ### 1. 범위 내 마이그레이션 찾기 - 브랜치 또는 PR을 검토하는 경우: `git diff --name-only origin/main...HEAD` 및 마이그레이션 폴더(`migrations/`, `db/migrate/`, `alembic/versions/`, `prisma/migrations/`, `database/migrations/`, `db/migration/`) 아래의 파일 유지. - 그렇지 않으면 사용자가 가리킨 파일 또는 SQL을 사용하세요. - 어떤 마이그레이션이 새 것인지 이미 어떤 환경에 적용되었는지 기록하세요. 적용된 마이그레이션을 편집하는 것을 제안하지 마세요; 새 후속 마이그레이션을 제안하세요. ### 2. 컨텍스트 설정 구성 파일, docker-compose, 또는 사용자에게 물어서 다음을 결정하세요: - 엔진 및 주요 버전(예: PostgreSQL 15, MySQL 8.0). 잠금 동작은 이에 따라 달라집니다. - 각 테이블의 대략적인 크기와 쓰기 트래픽. - 배포 방식: 마이그레이션이 새 코드 롤아웃 전, 중, 후에 실행됩니까? 크기나 트래픽을 모르면, 테이블이 크고 활발하다고 가정하고 그렇게 말하세요. ### 3. 실제 SQL 얻기 ORM 코드는 실제로 실행되는 것을 숨깁니다. 먼저 SQL을 렌더링하세요: | 프레임워크 | 명령 | |-----------|---------| | Django | `python manage.py sqlmigrate <app> <migration>` | | Rails | 스크래치 DB에서 `rails db:migrate` 실행, 그런 다음 `db/structure.sql` diff 검사 | | Alembic | `alembic upgrade <from>:<to> --sql` | | Prisma | `prisma/migrations/<name>/migration.sql` 읽기 | | Laravel | `php artisan migrate --pretend` | | Knex | `DEBUG=knex:query`로 스크래치 DB에서 실행하고 기록된 SQL 복사 | | Flyway / Liquibase | `.sql` 파일 또는 `liquibase update-sql` | 렌더링된 SQL을 이미 `.sql` 파일이 아니면 임시 파일에 저장하세요. ### 4. 스캐너 실행 ```bash python3 scripts/scan_migration.py --dialect postgres path/to/migration.sql python3 scripts/scan_migration.py --dialect mysql db/*.sql ``` `file:line [SEVERITY] RULE message`를 출력하고 HIGH 발견이 있으면 종료 코드 1을 반환합니다. 출력을 판정이 아닌 단서로 취급하세요: 정규식을 사용하며, 동적 SQL을 놓칠 수 있고, 테이블 크기를 알 수 없습니다. ### 5. 모든 문장 수동 검토 각 문장에 대해 `references/risk-catalog.md`를 사용하여 답하세요: 1. 어떤 잠금을 취하며, 얼마나 오래(즉시, 테이블 스캔, 또는 전체 재작성)? 2. 데이터를 잃거나 손상시킬 수 있나요? 의도된 것이며 백업이 있나요? 3. 긴 트랜잭션 뒤에 대기열에 쌓이나요? `lock_timeout`(Postgres) 또는 `lock_wait_timeout`(MySQL)이 설정되어 모든 트래픽을 차단하는 대신 빠르게 실패하나요? 4. 트랜잭션에서 실행되어서는 안 되는 곳에서 실행되나요(예: `CREATE INDEX CONCURRENTLY`)? 5. 대규모 데이터 백필이 일괄 처리되고 DDL과 분리되어 있나요? ### 6. 애플리케이션 호환성 확인 롤링 배포 중에, 기존 및 새 코드가 새 스키마에 대해 동시에 실행됩니다. `references/expand-contract.md`를 따르세요: - 모든 이름 변경, 삭제, 또는 유형 변경 객체에 대해 코드베이스(`rg -n '<column_or_table_name>'`)를 검색하세요, 원시 SQL, 직렬화기, 및 분석 쿼리 포함. - 현재 배포된 코드가 견딜 수 없는 변경을 플래그하세요. ### 7. 롤백 경로 확인 - 다운 마이그레이션이 존재하며, 이전 상태를 실제로 복원합니까? - 삭제 및 손실 유형 변경은 일방통행입니다: 백업 또는 단계적 계획 필요. ### 8. 보고서 작성 `templates/review-report.md`를 정확히 채우세요. `examples/example-review.md`의 깊이와 일치하세요. 모든 HIGH 또는 MEDIUM 발견에 대해, 동일한 최종 상태를 안전하게 달성하는 대체 SQL 또는 마이그레이션 코드를 제공하고, 필요시 순서가 있는 배포 단계로 나누세요. ## 판정 - **SAFE** - 대형 테이블에 차단 잠금 없음, 데이터 손실 없음, 이전 버전과 호환. - **SAFE WITH CHANGES** - 나열된 재작성이 적용되면 배포 가능. - **UNSAFE** - 다운타임, 데이터 손실, 또는 실행 중인 코드 오류를 초래할 것. ## 규칙 - 프로덕션 또는 공유 데이터베이스에 대해 직접 마이그레이션을 실행하지 마세요. - 사용자가 요청하지 않는 한 마이그레이션 파일을 수정하지 마세요; 보고서에서 변경 사항을 제안하세요. - 구체적으로: 테이블, 잠금, 및 실패 모드를 명명하세요. 일반적인 조언을 건너뛰세요. - 버전별 동작에 대해 확실하지 않으면, 그렇게 말하고 프로덕션 크기 복사본에서 `\timing` / `EXPLAIN` 및 잠금 모니터링으로 테스트를 제안하세요. FILE:references/risk-catalog.md # 위험 카탈로그: 일반적인 마이그레이션 작업 잠금 이름은 PostgreSQL입니다. ACCESS EXCLUSIVE는 모든 읽기 및 쓰기를 차단합니다; SHARE는 쓰기를 차단합니다; SHARE UPDATE EXCLUSIVE는 둘 다 차단하지 않습니다. ## 잠금 대기열 문제(아래 모든 것에 적용) "즉시" ALTER TABLE조차도 잠시 ACCESS EXCLUSIVE가 필요합니다. 긴 쿼리 또는 유휴 트랜잭션 세션이 테이블을 보유하면, ALTER는 대기합니다 - 그리고 모든 새 쿼리가 그 뒤에 대기열을 형성합니다. 1ms 변경이 여러 분 다운타임을 초래할 수 있습니다. 항상 위험한 마이그레이션을 다음으로 시작하세요: ```sql SET lock_timeout = '5s'; -- 빠르게 실패, 나중에 재시도 SET statement_timeout = '15min'; -- 선택적 상한 ``` MySQL 동등: `SET SESSION lock_wait_timeout = 5;` (메타데이터 잠금. ## PostgreSQL 작업 | 작업 | 위험 | 안전한 패턴 | |-----------|------|--------------| | `CREATE INDEX` | SHARE 잠금: 전체 빌드 동안 쓰기 차단 | `CREATE INDEX CONCURRENTLY`, 트랜잭션 외부; 실패 시 INVALID 인덱스 삭제 및 재시도. Rails: `disable_ddl_transaction!`; Django: `atomic = False` | | `DROP INDEX` | ACCESS EXCLUSIVE | `DROP INDEX CONCURRENTLY` | | `ADD COLUMN` nullable, 기본값 없음 | 즉시 | 안전(여전히 lock_timeout 설정) | | `ADD COLUMN ... DEFAULT <constant>` | PG 11+에서 즉시, 11 이전 재작성 | 11+에서 안전 | | `ADD COLUMN ... DEFAULT now()/random()/gen_random_uuid()` | 휘발성 기본값: 전체 테이블 재작성 | nullable 열 추가, 일괄 백필, 그런 다음 기본값 설정 | | `ADD COLUMN ... NOT NULL` 기본값 없음 | 비어 있지 않은 테이블에서 실패 | nullable 추가, 백필, 그런 다음 NOT NULL 적용(아래) | | `ALTER COLUMN ... SET NOT NULL` | ACCESS EXCLUSIVE 하에 전체 스캔 | `ADD CONSTRAINT c CHECK (col IS NOT NULL) NOT VALID`; `VALIDATE CONSTRAINT c`; 그런 다음 `SET NOT NULL` (PG 12+는 스캔 건너뜀); `c` 삭제 | | `ALTER COLUMN ... TYPE` | 일반적으로 전체 재작성 + ACCESS EXCLUSIVE 하에 인덱스 재구축 | 이진 강제 변환 가능한 경우에만 안전(varchar(n)에서 더 큰 n 또는 text로). 그렇지 않으면 새 열 + 이중 쓰기 + 백필 + 교체 | | `ADD FOREIGN KEY` | 모든 행 검증하는 동안 두 테이블 잠금 | `ADD CONSTRAINT ... NOT VALID`, 그런 다음 별도 단계에서 `VALIDATE CONSTRAINT` | | `ADD CHECK` | ACCESS EXCLUSIVE 하에 스캔 | 동일한 NOT VALID + VALIDATE 패턴 | | `ADD UNIQUE` / `ADD PRIMARY KEY` | 잠금 하에 인덱스 구축 | `CREATE UNIQUE INDEX CONCURRENTLY idx ...`; 그런 다음 `ADD CONSTRAINT ... UNIQUE USING INDEX idx` | | `RENAME COLUMN` / `RENAME TO` | 즉시, 그러나 실행 중인 코드 중단 | 확장/축소(expand-contract.md 참조) | | `DROP COLUMN` | 즉시, 그러나 되돌릴 수 없음; 선택하는 기존 코드 오류 | 모든 코드 참조 제거 및 먼저 배포; 그런 다음 삭제 | | `DROP TABLE` / `TRUNCATE` | 되돌릴 수 없는 데이터 손실 | 백업 및 제로 독자 확인; 먼저 `_deprecated`로 이름 변경 고려 | | `ALTER TYPE ... ADD VALUE` | 새 값은 같은 트랜잭션에서 사용 불가; PG 12 이전에는 트랜잭션 전혀 없음 | 자체 마이그레이션에 넣기 | | `VACUUM FULL` / `CLUSTER` / `REINDEX` | ACCESS EXCLUSIVE 하에 전체 재작성 | `REINDEX CONCURRENTLY` (PG 12+), 블로트에는 `pg_repack` | | 큰 `UPDATE` / `DELETE` | 긴 행 잠금, WAL 스파이크, 복제 지연 | 기본 키로 일괄 처리(1k-10k 행), 일괄당 커밋, DDL 마이그레이션 외부에서 실행 | ## MySQL 8.0 (InnoDB) 참고 사항 - MySQL이 조용히 테이블을 복사하는 대신 오류를 내도록 항상 알고리즘을 명시하세요: `ALTER TABLE t ADD COLUMN c INT, ALGORITHM=INSTANT;` 또는 `ALTER TABLE t ADD INDEX i (c), ALGORITHM=INPLACE, LOCK=NONE;` - `ADD COLUMN`은 8.0.12+ (마지막 위치) 및 8.0.29+ (모든 위치)에서 INSTANT입니다. - `MODIFY` / `CHANGE COLUMN` 유형 변경은 ALGORITHM=COPY를 사용합니다: 쓰기 차단. - COPY 전용 변경이 있는 큰 테이블의 경우 `gh-ost` 또는 `pt-online-schema-change`를 사용하세요. - MySQL에서 DDL은 트랜잭션이 아닙니다: 실패한 다중 문장 마이그레이션은 스키마가 절반 적용된 상태로 남깁니다. 마이그레이션당 하나의 DDL 문장을 유지하세요. FILE:references/expand-contract.md # 확장 / 축소: 이전 버전과 호환되는 스키마 변경 롤링 배포 중에, 기존 및 새 애플리케이션 버전이 나란히 실행됩니다. 마이그레이션이 새 코드가 라이브되기 전에 실행되면, 기존 코드는 새 스키마에서 작동해야 합니다. 실행 후라면, 새 코드는 기존 스키마에서 작동해야 합니다. 확장/축소는 모든 단계를 둘 다와 호환되게 만듭니다. ## 세 단계 1. **확장** - 새 구조만 추가(열, 테이블, 인덱스). 아무것도 제거되거나 이름 변경되지 않습니다. 기존 코드는 추가를 무시합니다. 2. **마이그레이션** - 기존 및 새 구조 모두에 쓰는 코드를 배포하고, 기존 행을 일괄 백필한 다음, 새 구조로 읽기를 전환합니다. 3. **축소** - 배포된 코드가 기존 구조를 건드리지 않으면, 별도의 나중 마이그레이션에서 삭제합니다. 각 단계는 자체 배포입니다. 하나의 마이그레이션에서 확장과 축소를 결합하지 마세요. ## 레시피 ### 열 이름 변경(`users.name`을 `users.full_name`로) 1. 마이그레이션: nullable `full_name` 추가. 2. 코드: `name`과 `full_name` 모두에 쓰기; `name` 읽기. 3. `full_name IS NULL`인 곳에서 `full_name = name`을 일괄 백필. 4. 코드: `full_name` 읽기; 둘 다에 계속 쓰기. 5. 코드: `name` 쓰기 중지. (Rails: 여기서 `name`을 `ignored_columns`에 추가.) 6. 마이그레이션: `name` 삭제. ### 열 유형 변경(`orders.amount` int를 numeric으로) 이름 변경과 동일: `amount_numeric` 추가, 이중 쓰기, 백필, 읽기 전환, 기존 삭제. 애플리케이션 변경이 어려우면 트리거가 이중 쓰기를 처리할 수 있습니다. ### 열을 NOT NULL로 만들기 1. 코드: 항상 값을 쓰기. 2. NULL 행을 일괄 백필. 3. 마이그레이션: CHECK ... NOT VALID, VALIDATE, SET NOT NULL (risk-catalog.md 참조). ### 열 또는 테이블 삭제 1. 코드: 모든 읽기 및 쓰기 제거(ORM 모델, 원시 SQL, 뷰, 보고서, ETL 작업, 및 데이터베이스를 공유하는 다른 서비스 검색). 2. 배포 및 전체 릴리스 주기 이상 대기. 3. 마이그레이션: 삭제. 중요하면 먼저 데이터 백업 또는 스냅샷 찍기. ### 테이블 분할 또는 이동 새 테이블 생성, 이중 쓰기, 백필, 읽기 전환, 기존 쓰기 중지, 삭제. ## 각 변경에 대해 답해야 할 호환성 질문 - 배포된 코드가 `SELECT *`를 하거나 모든 열을 매핑합니까(ORMs는 종종 부팅 시 열 목록을 캐시하고 하나가 사라지면 실패)? - 기존 코드의 삽입이 기본값 없는 새 NOT NULL 열 때문에 실패합니까? - 다른 서비스, 크론 작업, BI 대시보드, 또는 복제본이 이 테이블을 읽습니까? - 다운 마이그레이션 없이 배포를 이전 코드 버전으로 롤백할 수 있습니까? 마지막 질문에 대한 답이 "아니오"이면, 변경은 이전 버전과 호환되지 않습니다. FILE:templates/review-report.md # 마이그레이션 안전성 검토: <마이그레이션 이름 또는 PR 제목> **판정:** SAFE | SAFE WITH CHANGES | UNSAFE **엔진:** <예: PostgreSQL 15> | **검토된 파일:** <개수> **가정:** <테이블 크기, 트래픽, 배포 순서 - 추측된 것 표시> ## 요약 <2-4 문장: 마이그레이션이 무엇을 하는지, 가장 큰 위험, 무엇을 변경해야 하는지.> ## 발견 사항 | # | 심각도 | 파일:줄 | 문장 | 위험 | |---|----------|-----------|-----------|------| | 1 | HIGH | <경로:줄> | `<짧은 SQL>` | <잠금 / 데이터 손실 / 기존 코드 중단> | ### 1. <발견 사항의 짧은 제목> - **무슨 일이:** <취한 잠금, 지속 시간, 차단된 대상, 또는 무엇이 중단되는지> - **여기서 왜 중요한지:** <테이블 크기, 트래픽, 의존하는 코드> - **안전한 대안:** ```sql -- 대체 SQL 또는 마이그레이션 코드, 실행 순서대로 ``` <각 HIGH 및 MEDIUM 발견 사항에 대해 반복. LOW 발견 사항은 하나의 목록으로 그룹화.> ## 애플리케이션 호환성 - <각 이름 변경 / 삭제 / 유형 변경 객체 및 코드가 여전히 사용하는 위치> - <또는: "영향받는 코드 참조 없음 - X, Y에 대해 rg로 확인."> ## 롤백 계획 - <다운 마이그레이션이 상태를 복원합니까? 되돌릴 수 없는 것은?> - <실행 전 백업 또는 스냅샷 필요: 예/아니오> ## 권장 배포 순서 1. <마이그레이션 또는 코드 배포 단계> 2. <...> ## 스캐너 출력 ``` <scripts/scan_migration.py 출력 붙여넣기, 또는 거짓 긍정 기록> ``` FILE:examples/example-review.md # 예시 검토 **입력:** PR "주문 상태 추적 추가" with 하나의 Rails 마이그레이션 for PostgreSQL 15. `orders`는 ~40M 행 및 지속적인 쓰기 수신. 마이그레이션은 새 코드 전에 실행. ```sql -- 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); ``` **스캐너:** 3 HIGH (이름 변경, 인덱스-비-동시, FK-검증됨), 1 MEDIUM (잠금-시간-초과-없음). --- # 마이그레이션 안전성 검토: 주문 상태 추적 추가 **판정:** UNSAFE **엔진:** PostgreSQL 15 | **검토된 파일:** 1 **가정:** orders ~40M 행, 높은 쓰기 트래픽 (사용자로부터); carriers는 작음. ## 요약 주문 상태 열 추가, `shipped_on` 이름 변경, 상태 인덱싱, 및 운송업체 외래 키 추가. 상태 열 자체는 PG 15에서 안전하지만, 이름 변경은 실행 중인 앱을 중단시키고, 인덱스 및 FK는 `orders`에 대한 쓰기를 몇 분간 차단합니다. 세 개의 마이그레이션으로 분할하고 동시 / NOT VALID 변형을 사용하세요. ## 발견 사항 | # | 심각도 | 파일:줄 | 문장 | 위험 | |---|----------|-----------|-----------|------| | 1 | HIGH | rendered.sql:3 | `RENAME COLUMN shipped_on` | 배포 시 기존 코드 오류 | | 2 | HIGH | rendered.sql:4 | `CREATE INDEX ... (status)` | 빌드 중 쓰기 차단 | | 3 | HIGH | rendered.sql:5 | `ADD ... FOREIGN KEY` | 잠금 하에 전체 검증 스캔 | | ㅁ4 | MEDIUM | rendered.sql:1 | `lock_timeout` 없음 | ALTERs가 대기열을 형성하고 트래픽을 지연시킬 수 있음 | ### 1. 열 이름 변경이 실행 중인 코드를 중단시킴 - **무슨 일이:** 이름 변경은 즉시지만, 앱 서버는 여전히 기존 릴리스에서 `shipped_on`을 쿼리하고 배포가 끝날 때까지 `column does not exist`로 실패합니다. - **여기서 왜 중요한지:** `rg -n shipped_on`은 7개의 참조를 찾습니다, `app/serializers/order_serializer.rb` 및 야간 `reports/fulfillment.sql` 포함. - **안전한 대안:** 확장/축소. `shipped_at` 추가, 이중 쓰기, 일괄 백필, 읽기 전환, 그런 다음 나중 릴리스에서 `shipped_on` 삭제. ### 2. 인덱스 빌드가 쓰기를 차단함 - **안전한 대안** (별도 마이그레이션, `disable_ddl_transaction!`): ```sql CREATE INDEX CONCURRENTLY index_orders_on_status ON orders (status); ``` ### 3. 외래 키가 잠금 하에 40M 행을 검증함 - **안전한 대안:** ```sql SET lock_timeout = '5s'; ALTER TABLE orders ADD CONSTRAINT fk_orders_carrier FOREIGN KEY (carrier_id) REFERENCES carriers (id) NOT VALID; -- 다음 마이그레이션 (orders에 SHARE UPDATE EXCLUSIVE만 소요): ALTER TABLE orders VALIDATE CONSTRAINT fk_orders_carrier; ``` **LOW:** 없음. `ADD COLUMN ... DEFAULT 'pending'`은 PG 11+에서 메타데이터 전용입니다. ## 애플리케이션 호환성 - `shipped_on`: 7 코드 참조 및 하나의 SQL 보고서; 축소 단계까지 유지되어야 함. ## 롤백 계획 - 다운 마이그레이션은 `status`를 삭제합니다 (데이터 손실 허용: 새 열). 이름 변경은 되돌릴 수 있습니다. - 이름 변경이 제거되면 이 변경 세트에 백업 필요 없음. ## 권장 배포 순서 1. 마이그레이션 A: `SET lock_timeout`; `status` 추가; `shipped_at` 추가; FK NOT VALID 추가. 2. 마이그레이션 B (트랜잭션 없음): 상태 인덱스를 동시에 생성. 3. 마이그레이션 C: FK 검증. `shipped_on`/`shipped_at` 이중 쓰기 코드 배포. 4. `shipped_at` 백필; 읽기 전환; 나중 릴리스에서 `shipped_on` 삭제. FILE:scripts/scan_migration.py #!/usr/bin/env python3 """마이그레이션 파일에서 위험한 SQL을 위한 휴리스틱 스캐너 (PostgreSQL / MySQL). 사용법: python3 scan_migration.py [--dialect postgres|mysql] FILE [FILE ...] 종료 코드: 0 = HIGH 발견 없음, 1 = HIGH 발견, 2 = 사용법 오류.""" import re, sys F = re.I | re.S COLDEF = r"(?:\([^)]*\)|[^,(])*" # 하나의 열 정의, numeric(10,2) 허용 RULES = [ # (심각도, 규칙 ID, 방언 또는 둘 다 None, 정규식, 메시지) ("HIGH", "drop-table", None, r"^DROP\s+TABLE\b", "되돌릴 수 없는 데이터 손실; 백업 및 제로 독자 확인"), ("HIGH", "truncate", None, r"^TRUNCATE\b", "되돌릴 수 없는 데이터 손실"), ("HIGH", "drop-column", None, r"^ALTER\s+TABLE\b.*\bDROP\s+(COLUMN\b|(?!CONSTRAINT|INDEX|KEY|PRIMARY|FOREIGN|CHECK|DEFAULT|NOT|IDENTITY|EXPRESSION)\w)", "데이터 손실; 읽는 배포 코드가 실패합니다 - 먼저 코드 참조 제거"), ("HIGH", "rename", None, r"^ALTER\s+TABLE\b.*\bRENAME\b", "실행 중인 코드 중단; 확장/축소 사용"), ("HIGH", "type-change", "postgres", r"^ALTER\s+TABLE\b.*\bALTER\s+(COLUMN\s+)?\S+\s+(SET\s+DATA\s+)?TYPE\b", "일반적으로 ACCESS EXCLUSIVE 하에 전체 테이블 재작성"), ("HIGH", "type-change", "mysql", r"^ALTER\s+TABLE\b.*\b(MODIFY|CHANGE)\s+(COLUMN\s+)?\S+", "열 재정의는 일반적으로 ALGORITHM=COPY를 사용합니다 (쓰기 차단)"), ("HIGH", "index-not-concurrent", "postgres", r"^CREATE\s+(UNIQUE\s+)?INDEX\s+(?!CONCURRENTLY)", "빌드 중 쓰기 차단; CREATE INDEX CONCURRENTLY 사용"), ("MEDIUM", "drop-index-not-concurrent", "postgres", r"^DROP\s+INDEX\s+(?!CONCURRENTLY)", "ACCESS EXCLUSIVE 소요; DROP INDEX CONCURRENTLY 사용"), ("HIGH", "fk-validated", "postgres", r"^ALTER\s+TABLE\b(?!.*\bNOT\s+VALID\b).*\b(FOREIGN\s+KEY|REFERENCES)\b", "두 테이블을 잠그면서 모든 행 검증; NOT VALID 추가, 그런 다음 VALIDATE"), ("MEDIUM", "check-validated", "postgres", r"^ALTER\s+TABLE\b(?!.*\bNOT\s+VALID\b).*\bADD\s+(CONSTRAINT\s+\S+\s+)?CHECK\b", "잠금 하에 전체 스캔; NOT VALID 추가, 그런 다음 VALIDATE"), ("MEDIUM", "set-not-null", "postgres", r"\bSET\s+NOT\s+NULL\b", "ACCESS EXCLUSIVE 하에 전체 스캔; 먼저 CHECK (col IS NOT NULL) 검증"), ("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", "비어 있지 않은 테이블에서 실패 (또는 기존 코드 삽입 실패); nullable 추가, 백필, 그런 다음 적용"), ("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*\(", "휘발성 기본값이 테이블 재작성; nullable 추가, 백필, 그런 다음 기본값 설정"), ("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", "잠금 하에 인덱스 구축; CONCURRENTLY로 생성, 그런 다음 ADD CONSTRAINT ... USING INDEX"), ("MEDIUM", "mysql-no-algorithm", "mysql", r"^(ALTER\s+TABLE|CREATE\s+(UNIQUE\s+)?INDEX)\b(?!.*\bALGORITHM\s*=)", "ALGORITHM=INSTANT|INPLACE, LOCK=NONE 명시하여 MySQL이 차단 복사를 거부하도록"), ("HIGH", "dml-no-where", None, r"^(UPDATE|DELETE)\b(?!.*\bWHERE\b)", "하나의 트랜잭션에서 모든 행 접촉; 일괄 처리"), ("LOW", "dml-in-migration", None, r"^(UPDATE|DELETE|INSERT)\b.*\bWHERE\b", "마이그레이션에서 데이터 변경; 테이블이 크면 일괄 처리"), ("MEDIUM", "table-rewrite", "postgres", r"^(VACUUM\s+FULL|CLUSTER|REINDEX\s+(?!.*CONCURRENTLY))", "ACCESS EXCLUSIVE 하에 재작성; REINDEX CONCURRENTLY 또는 pg_repack 사용"), ("LOW", "enum-add-value", "postgres", r"^ALTER\s+TYPE\b.*\bADD\s+VALUE\b", "새 값은 같은 트랜잭션에서 사용 불가; 자체 마이그레이션에 유지"), ] def statements(sql): """주석 제거 후 (줄 번호, 문장) 생성. 단순 ';' 분할.""" 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:]))))