Feature/query refactor

ER_DOSE_ERROR.md

@@ -8,7 +8,7 @@
 mbeat.er_data_raw
   -> er_dose batch
   -> prism_common.er_dose_raw_parsed
-  -> prism_common.de_trend_die_yield_daily (summary)
+  -> Airflow 후속 작업: prism_common.de_trend_die_yield_daily (summary)
 ```
 
 Root cause는 이 배치와 별도 흐름이다.
@@ -17,7 +17,7 @@ Root cause는 이 배치와 별도 흐름이다.
 mbeat.er_data_raw_euv
   -> contents root cause 파싱
   -> prism_common.er_dose_euv_parsed
-  -> prism_common.de_trend_root_cause_daily (summary)
+  -> Airflow 후속 작업: prism_common.de_trend_root_cause_daily (summary)
 ```
 
 `prism_common.er_dose_raw_parsed`와 `prism_common.er_dose_euv_parsed`는 서로 조인하거나 매칭하지 않는다.
@@ -30,6 +30,7 @@ mbeat.er_data_raw_euv
 - `mbeat.er_data_raw_euv`: Root cause source description 후보 RAW. `contents`에 `dose error detected in file`, `root cause`, `exposure id`, 각종 EUV 지표가 들어온다. `er_date`, `er_index`가 없다.
 - `prism_common.er_dose_euv_parsed`: FE 조회용 root cause 결과 테이블. `er_data_raw_euv.contents`를 파싱한 구조화 컬럼과 원문을 저장하며, `er_dose_raw_parsed`와 무관하다.
 - `prism_common.de_trend_root_cause_daily`: EUV Root Cause 일별 발생 빈도 요약 서머리 테이블 (`occur_date`, `eq_name`, `root_cause`, `frequency`).
+- `mbeat.batch_event_log`: 여러 배치가 공통으로 사용하는 이벤트 로그 테이블. `batch_name`, `target_date`, `event_type`, `message`와 가변 상세 데이터인 `data jsonb`를 저장한다.
 
 DDL:
 
@@ -184,22 +185,23 @@ Mermaid ERD는 렌더링 호환성을 위해 타입 표기를 단순화했다. 
    - RAW는 이전 chunk의 `COPY`를 적재 worker 1개에서 실행하는 동안 다음 chunk를 조회·파싱한다. 동시에 대기하는 적재 작업은 1개로 제한해 처리 순서와 메모리 사용량을 유지한다.
 4. 각 `chunk`를 `prism_common.er_dose_raw_parsed` 일별 파티션에 `COPY` append insert
    - 파티션 적재는 공통 `copy_insert_to_partition_table`을 사용하며, DataFrame 컬럼을 테이블 물리 컬럼 순서와 동일하게 정렬한 뒤 `COPY ... FROM STDIN WITH CSV HEADER`를 실행한다.
-5. 적재된 파티션 날짜를 기준으로 DIE Yield 서머리 테이블(`prism_common.de_trend_die_yield_daily`) 및 EUV Root Cause 서머리 테이블(`prism_common.de_trend_root_cause_daily`)에 `UPSERT` 집계 업데이트 실행
+5. 적재한 파티션을 ANALYZE하고 파서 작업 종료
 
-환경변수 기반 기본 실행에서 target date와 `ER_DOSE_START_TIME`, `ER_DOSE_END_TIME`가 모두 없으면 raw/euv 배치는 최근 2일 lookback 모드로 동작한다.
+최근 날짜별 판단은 Airflow DAG에서 수행한다.
 
-1. 실행일 기준 `오늘 포함 최근 2일`을 날짜 오름차순으로 순회
-2. 각 날짜에 대해 원천 raw 전체 건수와 타겟 parsed 전체 건수를 비교
-3. 건수가 같으면 해당 날짜는 스킵
-4. RAW/EUV 건수가 다르고 parsed 건수가 0보다 크면 원천의 `eq_name`, `code`, `code_occur_time` 기준 중복 제거 건수를 추가 계산
-5. 중복 제거 건수가 parsed 건수와 같으면 스킵
-6. 중복 제거 건수도 다르거나 parsed 건수가 0이면 해당 날짜의 parsed 파티션을 `TRUNCATE`
-7. 원천 raw를 해당 날짜 처음부터 다시 조회해 chunk 단위로 파싱 후 insert
-8. 적재 완료 후 해당 날짜의 서머리 테이블 2종을 `UPSERT` 업데이트
+1. 기준일 포함 최근 10일을 날짜 오름차순으로 확인
+2. 각 날짜의 원천 raw 전체 건수와 parsed 전체 건수를 비교
+3. 건수가 같으면 해당 파서의 처리 대상에서 제외
+4. 전체 건수가 다르고 parsed 건수가 0보다 크면 원천의 `eq_name`, `code`, `code_occur_time` 기준 DISTINCT 건수를 추가 비교
+5. DISTINCT 건수도 다르거나 parsed 건수가 0이면 재적재 대상으로 확정
+6. 전체 날짜 계획을 XCom에 저장한 후, 날짜별 EUV → RAW 파서 작업 실행
+7. 대상 파서는 지정된 날짜 파티션을 TRUNCATE하고 재적재. 재시도에도 계획을 유지하며 건수를 다시 비교하지 않음
+8. 해당 날짜 적재 성공 후 서머리 두 종류와 RAW/EUV 통계 두 종류를 독립 Airflow 작업으로 실행
+9. 실패한 서머리·통계는 같은 기간으로 해당 작업만 재시도
 
-`ER_DOSE_EUV_TARGET_DATE`가 있으면 해당 날짜 1일만 같은 방식으로 count 비교 후 필요 시 재적재한다. `ER_DOSE_START_TIME`/`ER_DOSE_END_TIME`으로 시간 범위를 직접 지정하면 count 비교 없이 해당 범위를 처리한다.
+환경변수/CLI 파서 실행에는 날짜 또는 시간 범위를 반드시 지정한다. target date가 있으면 해당 날짜만 재적재하며, `ER_DOSE_START_TIME`/`ER_DOSE_END_TIME`으로 시간 범위를 직접 지정하면 기존 방식으로 해당 범위를 처리한다. 파서 단독 실행은 서머리·통계를 갱신하지 않는다.
 
-`ER_DOSE_EUV` 배치는 `mbeat.er_data_raw_euv`를 기간 조건으로 `chunk` 조회하고, root cause 형식의 `contents`만 파싱해 `prism_common.er_dose_euv_parsed`에 적재한다. EUV source count도 parsed count와 맞추기 위해 `contents`에 `dose error detected in file:`과 `root cause`가 있는 row만 계산하며, 전체 건수가 다를 때는 RAW와 동일하게 고유 이벤트 count를 추가 비교한다. EUV parsed 결과에는 `eq_name`, `er_type`, `code`, `code_occur_time`, `title`, `contents`, `reason_code`, `task`, `compile_script`와 root cause 파싱 컬럼만 저장하며, 적재 완료 후 `de_trend_root_cause_daily` 서머리 테이블을 `UPSERT` 업데이트한다.
+`ER_DOSE_EUV` 배치는 `mbeat.er_data_raw_euv`를 기간 조건으로 `chunk` 조회하고, root cause 형식의 `contents`만 파싱해 `prism_common.er_dose_euv_parsed`에 적재한다. EUV source count도 parsed count와 맞추기 위해 `contents`에 `dose error detected in file:`과 `root cause`가 있는 row만 계산하며, 전체 건수가 다를 때는 RAW와 동일하게 고유 이벤트 count를 추가 비교한다. EUV parsed 결과에는 `eq_name`, `er_type`, `code`, `code_occur_time`, `title`, `contents`, `reason_code`, `task`, `compile_script`와 root cause 파싱 컬럼만 저장하며, Root Cause 집계와 건수 로그는 이후 RAW 마지막 단계에서 저장한다.
 RAW와 EUV 모두 대용량 처리를 위해 전체 결과를 한 번에 메모리로 올리지 않고 `read chunk -> parse -> insert` 방식으로 반복 처리한다.
 또한, 데이터베이스 드라이버 단의 메모리 팽창을 방지하기 위해 SQLAlchemy 서버사이드 커서(`stream_results=True`, `max_row_buffer=chunk_size`)를 활성화하여 스트리밍 조회를 수행한다. 다만 실제 메모리 사용량은 `chunk` 크기와 raw `contents` 크기에 영향을 받기 때문에 운영 환경에서 조정이 필요할 수 있다.
 RAW와 EUV 모두 조회 SQL에서 `prism_dev.photo_eqp_info`의 `use_yn = 'Y'`이고 `eqp_model_name like 'NXE%'`인 `eqp_id`를 서브쿼리로 조회해 `eq_name` 필터로 사용한다. RAW의 이전 `lot_seq`, `wafer_seq` 상태 조회에도 같은 조건을 적용한다.
@@ -258,10 +260,12 @@ RAW parsed 저장 필드:
 
 ## LO-0050/LO-0051/LO-0052 파싱 규칙
 
-- `LO-0050`의 lot 시작, `LO-0051`의 lot 종료, `LO-0052`의 lot 중단 메시지에 동일한 파싱 규칙을 적용한다.
-- `lot_id`: 원문 텍스트의 `lot '([^']+)'` 정규식 패턴에서 추출한다.
+- `LO-0050`의 lot 시작, `LO-0051`의 lot 종료, `LO-0052`의 lot 중단 메시지에서 동일한 lot 필드를 저장한다.
+- `LO-0050/LO-0051`은 기존 `lot '...' (id=...)` 형식을 사용한다.
+- `LO-0052`는 별도 코드 분기에서 `lot '...' (id=...)`와 `lot (name='...', id=...)` 형식을 모두 처리한다.
+- `lot_id`: 각 메시지의 lot 문자열에서 추출한다.
 - `lot_name`: 추출된 `lot_id`에서 첫 번째 `.`(점) 문자를 기준으로 이전 텍스트를 추출(`lot_id.split('.', maxsplit=1)[0]`)한다.
-- `lot_seq`: `(id=\s*\d+)` 정규식 패턴에서 우선 추출하며, 미매칭 시 기존 `_LOT_SEQ_PATTERNS` 패턴으로 폴백한다.
+- `lot_seq`: 각 메시지의 `id` 값에서 추출하며, 미매칭 시 기존 `_LOT_SEQ_PATTERNS` 패턴으로 폴백한다.
 
 ## 실행
 
@@ -271,10 +275,11 @@ EUV 날짜 변수는 `ER_DOSE_EUV_TARGET_DATE` 를 사용한다.
 
 DB 접속은 `--dsn`, 프로젝트 루트 `er_dose.properties`, `ER_DOSE_DB_DSN`, `DATABASE_URL` 순서로 사용한다.
 기본 `chunk` 크기는 `ER_DOSE_RAW` 및 `ER_DOSE_EUV` 배치 모두 `30000`이며 `--chunk-size`로 조정할 수 있다.
-RAW 기본 실행은 최근 2일 lookback 모드이며, `--lookback-days` 또는 환경변수 기반 실행의 `ER_DOSE_LOOKBACK_DAYS`로 일수를 바꿀 수 있다.
+최근 10일 대상 결정은 Airflow DAG에서 수행한다. CLI에서는 `--date` 또는 `--start-time/--end-time`을 반드시 지정한다.
 
 ```bash
 python -m er_dose.run_er_dose_batch \
+  --date 2026-04-13 \
   --parser ER_DOSE_RAW \
   --chunk-size 30000 \
   --dsn 'postgresql://user:password@host:5432/dbname'
@@ -305,3 +310,17 @@ python -m er_dose.run_er_dose_batch \
   --chunk-size 30000 \
   --dsn 'postgresql://user:password@host:5432/dbname'
 ```
+
+## ER Dose 날짜별 Airflow 실행과 재시도
+
+`dags/er_dose_daily_dag.py`가 최근 10일의 건수를 비교해 날짜별 RAW/EUV 처리 여부를 결정합니다. 원천 건수와 parsed 건수가 같으면 제외하고, parsed 건수가 있을 때에는 기존과 동일하게 원천 DISTINCT 건수도 비교합니다. 계획은 `plan_dates` 작업의 XCom에 저장한 뒤 파싱을 시작합니다.
+
+파서는 `target_date` 또는 `start_time/end_time`을 명시적으로 받아 처리합니다. 파서 내부의 lookback과 서머리·통계 호출은 제거했습니다. `--date` 실행은 해당 날짜 파티션을 비우고 다시 적재하며, 시간 범위 직접 실행은 기존 적재 동작을 유지합니다. CLI 파서 실행만으로는 서머리·통계를 갱신하지 않습니다.
+
+DAG는 과거 날짜부터 EUV → RAW 순서로 적재하고, 해당 날짜 적재 성공 후 수율 서머리·원인 서머리·RAW 통계·EUV 통계를 별도 작업으로 실행합니다. 서머리 실패는 다른 통계나 다음 날짜 파싱의 선행 조건이 아닙니다. 실제 병렬 실행 수는 Airflow executor/pool 설정에 따릅니다.
+
+서머리 작업은 기존 DELETE 메서드와 INSERT 메서드를 직접 호출하며, 두 실행을 하나의 트랜잭션으로 묶지 않습니다. 0건 집계도 DELETE 후 처리합니다. 통계는 SELECT 후 동일 배치·날짜·이벤트·시간 범위 메시지에 해당하는 이전 스냅샷을 DELETE하고 INSERT합니다. 재시도 시 중복 행이 남지 않도록 하며, 같은 기간의 과거 스냅샷은 교체됩니다.
+
+RAW와 EUV 로그는 별도 행이며, `data`에는 `equipment_counts` 배열만 저장합니다. 각 원소는 `eq_name`, `source_count`, `target_count`입니다. 시간 범위는 `message`, 시작일은 `target_date`, 저장 시각은 `created_at`에 기록합니다. 메시지와 JSON 구성은 `airflow_modules/er_dose_jobs.py`에 있고, 공통 로그 repository는 전달받은 조건과 데이터만 처리합니다.
+
+DAG 배포·운영 및 장애 복구 절차는 [Airflow 재처리 안내](docs/er_dose_airflow.md)를 참고하세요. 이전 스레드 구성의 측정치는 [과거 검증 기록](docs/pr/er_dose_final_statistics_validation.md)에 있으며 새 DAG의 성능 측정치는 아닙니다.