⚡ Bolt: [performance improvement] job_store.py의 SQLite PRAGMA journal_mode=WAL 실행 최적화

.jules/bolt.md

@@ -1,3 +1,7 @@
+## 2024-08-16 - [Optimize SQLite WAL mode PRAGMA execution]
+**Learning:** SQLite의 `PRAGMA journal_mode=WAL` 설정은 데이터베이스 파일 당 영구적으로 유지되므로, 잦은 짧은 연결(short-lived connections)마다 반복 실행하면 불필요한 연산 오버헤드가 발생합니다.
+**Action:** 스키마 초기화 시 `conn.executescript()`를 통해 최초 1회만 설정하도록 변경하여 매 연결마다 PRAGMA를 호출하는 낭비를 줄입니다.
+
 ## 2024-05-28 - Avoid O(N^2) Path.resolve() in Batch Processing
 **Learning:** Python's `pathlib.Path.resolve()` is relatively slow because it touches the filesystem to follow symlinks and resolve relative paths. When dealing with a batch operation (e.g., scanning large directories of media files), calculating protected files via `any(target == src.resolve() for src in sources)` on every check leads to massive O(N^2) CPU overhead.
 **Action:** Pre-resolve the entire list of candidate paths once into a `frozenset` at the beginning of the batch process. Pass this resolved set down the call stack so that collision/protection checks become O(1) hash map lookups instead of triggering millions of unnecessary disk access operations.

job_store.py

@@ -95,7 +95,7 @@ def __init__(self, db_path: str) -> None:
         self._db_path = str(db_path)
         self._lock = threading.Lock()
         with self._connect() as conn:
-            conn.execute(_SCHEMA)
+            conn.executescript("PRAGMA journal_mode=WAL;\n" + _SCHEMA)
 
     @contextmanager
     def _connect(self) -> Iterator[sqlite3.Connection]:
@@ -108,7 +108,6 @@ def _connect(self) -> Iterator[sqlite3.Connection]:
         conn = sqlite3.connect(self._db_path, timeout=30.0)
         try:
             conn.row_factory = sqlite3.Row
-            conn.execute("PRAGMA journal_mode=WAL")
             yield conn
             conn.commit()
         finally:

run_benchmark.py

@@ -0,0 +1,13 @@
+import sqlite3
+import time
+import os
+
+from job_store import JobStore
+
+if os.path.exists("test_b.db"): os.remove("test_b.db")
+
+store_b = JobStore("test_b.db")
+start = time.time()
+for i in range(1000):
+    store_b.get("foo")
+print("Before:", time.time() - start)

test_pragma.py

@@ -0,0 +1,13 @@
+import sqlite3
+from job_store import JobStore
+import time
+import os
+
+if os.path.exists("test.db"):
+    os.remove("test.db")
+
+store = JobStore("test.db")
+start = time.time()
+for i in range(1000):
+    store.get("foo")
+print("Before optimization:", time.time() - start)

test_pragma2.py

@@ -0,0 +1,40 @@
+import sqlite3
+import time
+import os
+import contextlib
+
+class JobStore2:
+    def __init__(self, db_path: str) -> None:
+        self._db_path = str(db_path)
+        import threading
+        self._lock = threading.Lock()
+        with contextlib.closing(sqlite3.connect(self._db_path, timeout=30.0)) as conn:
+            conn.execute("PRAGMA journal_mode=WAL")
+            conn.executescript("CREATE TABLE IF NOT EXISTS jobs (id TEXT PRIMARY KEY, status TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, output_path TEXT, output_name TEXT, error TEXT, temp_dir TEXT)")
+
+    import contextlib as cl
+    @cl.contextmanager
+    def _connect(self):
+        conn = sqlite3.connect(self._db_path, timeout=30.0)
+        try:
+            conn.row_factory = sqlite3.Row
+            yield conn
+            conn.commit()
+        finally:
+            conn.close()
+
+    def get(self, job_id: str) -> dict | None:
+        with self._lock, self._connect() as conn:
+            row = conn.execute(
+                "SELECT * FROM jobs WHERE id = ?", (job_id,)
+            ).fetchone()
+        return dict(row) if row is not None else None
+
+if os.path.exists("test2.db"):
+    os.remove("test2.db")
+
+store = JobStore2("test2.db")
+start = time.time()
+for i in range(1000):
+    store.get("foo")
+print("After optimization:", time.time() - start)

test_pragma3.py

@@ -0,0 +1,74 @@
+import sqlite3
+import time
+import os
+import contextlib
+
+class JobStoreBefore:
+    def __init__(self, db_path: str) -> None:
+        self._db_path = str(db_path)
+        import threading
+        self._lock = threading.Lock()
+        with self._connect() as conn:
+            conn.execute("CREATE TABLE IF NOT EXISTS jobs (id TEXT PRIMARY KEY, status TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, output_path TEXT, output_name TEXT, error TEXT, temp_dir TEXT)")
+
+    import contextlib as cl
+    @cl.contextmanager
+    def _connect(self):
+        conn = sqlite3.connect(self._db_path, timeout=30.0)
+        try:
+            conn.row_factory = sqlite3.Row
+            conn.execute("PRAGMA journal_mode=WAL")
+            yield conn
+            conn.commit()
+        finally:
+            conn.close()
+
+    def get(self, job_id: str) -> dict | None:
+        with self._lock, self._connect() as conn:
+            row = conn.execute(
+                "SELECT * FROM jobs WHERE id = ?", (job_id,)
+            ).fetchone()
+        return dict(row) if row is not None else None
+
+
+class JobStoreAfter:
+    def __init__(self, db_path: str) -> None:
+        self._db_path = str(db_path)
+        import threading
+        self._lock = threading.Lock()
+        with self._connect() as conn:
+            conn.executescript("PRAGMA journal_mode=WAL;\nCREATE TABLE IF NOT EXISTS jobs (id TEXT PRIMARY KEY, status TEXT NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL, output_path TEXT, output_name TEXT, error TEXT, temp_dir TEXT)")
+
+    import contextlib as cl
+    @cl.contextmanager
+    def _connect(self):
+        conn = sqlite3.connect(self._db_path, timeout=30.0)
+        try:
+            conn.row_factory = sqlite3.Row
+            # Removed PRAGMA here
+            yield conn
+            conn.commit()
+        finally:
+            conn.close()
+
+    def get(self, job_id: str) -> dict | None:
+        with self._lock, self._connect() as conn:
+            row = conn.execute(
+                "SELECT * FROM jobs WHERE id = ?", (job_id,)
+            ).fetchone()
+        return dict(row) if row is not None else None
+
+if os.path.exists("test_b.db"): os.remove("test_b.db")
+if os.path.exists("test_a.db"): os.remove("test_a.db")
+
+store_b = JobStoreBefore("test_b.db")
+start = time.time()
+for i in range(10000):
+    store_b.get("foo")
+print("Before:", time.time() - start)
+
+store_a = JobStoreAfter("test_a.db")
+start = time.time()
+for i in range(10000):
+    store_a.get("foo")
+print("After:", time.time() - start)