ER Dose 서머리를 별도 DELETE·INSERT 메서드로 교체

Author: junho0831

er_dose/euv/euv_processor.py

@@ -118,7 +118,7 @@ def _run_window(
 
         for target_date in sorted(inserted_target_dates):
             self.repository.analyze_target_partition(target_date, connection=connection)
-            self.repository.upsert_root_cause_daily_summary(target_date, connection=connection)
+            self.repository.replace_root_cause_daily_summary(target_date)
             print(
                 "[ER_DOSE_EUV] "
                 f"summary updated (root_cause) partition_date={target_date}"

er_dose/euv/euv_repository.py

@@ -224,13 +224,25 @@ def analyze_target_partition(self, target_date: str, connection=None) -> int:
         partition_table = f"{ROOT_CAUSE_TABLE}_1_prt_p{target_date.replace('-', '')}"
         return self.db.execute(f"ANALYZE {partition_table}", connection=connection)
 
-    def upsert_root_cause_daily_summary(self, target_date: date | str, connection=None) -> int:
+    def replace_root_cause_daily_summary(self, target_date: date | str) -> None:
+        self.delete_root_cause_daily_summary(target_date)
+        self.insert_root_cause_daily_summary(target_date)
+
+    def delete_root_cause_daily_summary(self, target_date: date | str) -> None:
+        if isinstance(target_date, str):
+            target_date = datetime.strptime(target_date, "%Y-%m-%d").date()
+        delete_query = """
+            delete from prism_common.de_trend_root_cause_daily
+            where occur_date = :target_date
+        """
+        self.db.execute(delete_query, params={"target_date": target_date})
+
+    def insert_root_cause_daily_summary(self, target_date: date | str) -> None:
         if isinstance(target_date, str):
             target_date = datetime.strptime(target_date, "%Y-%m-%d").date()
         start_time = datetime.combine(target_date, datetime.min.time())
         end_time = start_time + timedelta(days=1)
-
-        query = f"""
+        insert_query = f"""
             insert into prism_common.de_trend_root_cause_daily (
                 occur_date,
                 eq_name,
@@ -274,13 +286,9 @@ def upsert_root_cause_daily_summary(self, target_date: date | str, connection=No
             from root_cause_data
             where root_cause is not null
               and root_cause != ''
-            group by occur_date, eq_name, root_cause
-            on conflict (occur_date, eq_name, root_cause)
-            do update set
-                frequency = excluded.frequency,
-                created_at = now();
+            group by occur_date, eq_name, root_cause;
         """
-        return self.db.execute(query, params={"start_time": start_time, "end_time": end_time}, connection=connection)
+        self.db.execute(insert_query, params={"start_time": start_time, "end_time": end_time})
 
     def transaction(self):
         return self.db.transaction()

er_dose/raw/raw_processor.py

@@ -165,8 +165,8 @@ def _run_window(
 
         for target_date in sorted(inserted_target_dates):
             self.repository.analyze_target_partition(target_date, connection=connection)
-            self.repository.upsert_die_yield_daily_summary(target_date, connection=connection)
-            self.repository.upsert_root_cause_daily_summary(target_date, connection=connection)
+            self.repository.replace_die_yield_daily_summary(target_date)
+            self.repository.replace_root_cause_daily_summary(target_date)
             print(
                 "[ER_DOSE] "
                 f"summary updated (die_yield, root_cause) partition_date={target_date}"
@@ -385,7 +385,7 @@ def _run_window(
 
         for target_date in sorted(inserted_target_dates):
             self.repository.analyze_target_partition(target_date, connection=connection)
-            self.repository.upsert_root_cause_daily_summary(target_date, connection=connection)
+            self.repository.replace_root_cause_daily_summary(target_date)
             print(
                 "[ER_DOSE_EUV] "
                 f"summary updated (root_cause) partition_date={target_date}"

er_dose/raw/raw_repository.py

@@ -232,13 +232,25 @@ def analyze_target_partition(self, target_date: str, connection=None) -> int:
         partition_table = f"{PARSED_TABLE}_1_prt_p{target_date.replace('-', '')}"
         return self.db.execute(f"ANALYZE {partition_table}", connection=connection)
 
-    def upsert_die_yield_daily_summary(self, target_date: date | str, connection=None) -> int:
+    def replace_die_yield_daily_summary(self, target_date: date | str) -> None:
+        self.delete_die_yield_daily_summary(target_date)
+        self.insert_die_yield_daily_summary(target_date)
+
+    def delete_die_yield_daily_summary(self, target_date: date | str) -> None:
+        if isinstance(target_date, str):
+            target_date = datetime.strptime(target_date, "%Y-%m-%d").date()
+        delete_query = """
+            delete from prism_common.de_trend_die_yield_daily
+            where occur_date = :target_date
+        """
+        self.db.execute(delete_query, params={"target_date": target_date})
+
+    def insert_die_yield_daily_summary(self, target_date: date | str) -> None:
         if isinstance(target_date, str):
             target_date = datetime.strptime(target_date, "%Y-%m-%d").date()
         start_time = datetime.combine(target_date, datetime.min.time())
         end_time = start_time + timedelta(days=1)
-
-        query = f"""
+        insert_query = f"""
             insert into prism_common.de_trend_die_yield_daily (
                 occur_date,
                 eq_name,
@@ -307,26 +319,29 @@ def upsert_die_yield_daily_summary(self, target_date: date | str, connection=Non
                 sum(case when valid_wafer_yn = 1 then reject_yn else 0 end) as reject_wafer,
                 now() as created_at
             from wafer_code_data
-            group by occur_date, eq_name
-            on conflict (occur_date, eq_name)
-            do update set
-                total_die = excluded.total_die,
-                reject_shot = excluded.reject_shot,
-                to_repair_die = excluded.to_repair_die,
-                repair_nok = excluded.repair_nok,
-                total_wafer = excluded.total_wafer,
-                reject_wafer = excluded.reject_wafer,
-                created_at = now();
+            group by occur_date, eq_name;
         """
-        return self.db.execute(query, params={"start_time": start_time, "end_time": end_time}, connection=connection)
+        self.db.execute(insert_query, params={"start_time": start_time, "end_time": end_time})
 
-    def upsert_root_cause_daily_summary(self, target_date: date | str, connection=None) -> int:
+    def replace_root_cause_daily_summary(self, target_date: date | str) -> None:
+        self.delete_root_cause_daily_summary(target_date)
+        self.insert_root_cause_daily_summary(target_date)
+
+    def delete_root_cause_daily_summary(self, target_date: date | str) -> None:
+        if isinstance(target_date, str):
+            target_date = datetime.strptime(target_date, "%Y-%m-%d").date()
+        delete_query = """
+            delete from prism_common.de_trend_root_cause_daily
+            where occur_date = :target_date
+        """
+        self.db.execute(delete_query, params={"target_date": target_date})
+
+    def insert_root_cause_daily_summary(self, target_date: date | str) -> None:
         if isinstance(target_date, str):
             target_date = datetime.strptime(target_date, "%Y-%m-%d").date()
         start_time = datetime.combine(target_date, datetime.min.time())
         end_time = start_time + timedelta(days=1)
-
-        query = """
+        insert_query = """
             insert into prism_common.de_trend_root_cause_daily (
                 occur_date,
                 eq_name,
@@ -345,13 +360,9 @@ def upsert_root_cause_daily_summary(self, target_date: date | str, connection=No
               and p.code_occur_time < :end_time
               and p.eq_name is not null
               and p.root_cause is not null
-            group by p.code_occur_time::date, p.eq_name, p.root_cause
-            on conflict (occur_date, eq_name, root_cause)
-            do update set
-                frequency = excluded.frequency,
-                created_at = now();
+            group by p.code_occur_time::date, p.eq_name, p.root_cause;
         """
-        return self.db.execute(query, params={"start_time": start_time, "end_time": end_time}, connection=connection)
+        self.db.execute(insert_query, params={"start_time": start_time, "end_time": end_time})
 
     def transaction(self):
         return self.db.transaction()

tests/test_er_dose_processor.py

@@ -282,7 +282,10 @@ def test_run_inserts_rows_without_deleting_existing_history(self):
         with redirect_stdout(StringIO()):
             processor.run(start_time=datetime(2026, 5, 1), end_time=datetime(2026, 5, 2))
 
-        delete_queries = [query for query, _, _ in db.executed if query.strip().lower().startswith("delete")]
+        delete_queries = [
+            query for query, _, _ in db.executed
+            if query.strip().lower().startswith("delete") and "er_dose_raw_parsed" in query.lower()
+        ]
         self.assertEqual(delete_queries, [])
         parsed_insert = self._inserted_df(db, "prism_common.er_dose_raw_parsed")
         self.assertNotIn("parser_version", parsed_insert.columns)

tests/test_er_dose_summary_replacement.py

@@ -0,0 +1,29 @@
+from datetime import date
+from unittest.mock import Mock
+import pytest
+from er_dose.raw.raw_repository import ERDoseRepository
+from er_dose.euv.euv_repository import ERDoseEUVRepository
+
+@pytest.mark.parametrize("repository_type,kind", [(ERDoseRepository,"die_yield"),(ERDoseRepository,"root_cause"),(ERDoseEUVRepository,"root_cause")])
+@pytest.mark.parametrize("failure", [None,"delete","insert"])
+def test_summary_uses_separate_statements(repository_type, kind, failure):
+    db = Mock()
+    events = []
+    def execute(query, params):
+        action = query.strip().split()[0].lower()
+        events.append(action)
+        if action == failure:
+            raise RuntimeError(action)
+    db.execute.side_effect = execute
+    repo = repository_type(db)
+    method = getattr(repo, 'replace_' + kind + '_daily_summary')
+    if failure:
+        with pytest.raises(RuntimeError, match=failure):
+            method(date(2026,5,1))
+    else:
+        assert method(date(2026,5,1)) is None
+    assert events == (['delete'] if failure == 'delete' else ['delete','insert'])
+    db.transaction.assert_not_called()
+    for call in db.execute.call_args_list:
+        assert 'connection' not in call.kwargs
+        assert 'on conflict' not in call.args[0].lower()