Use indexed node lookup sidecar

.env.example

@@ -4,6 +4,7 @@ MUSPROT_DB_FILENAME=MuSProt.db
 
 # Optional mounted-volume overrides
 # MUSPROT_DB_PATH=/data/MuSProt.db
+# MUSPROT_NODE_DB_PATH=/data/MuSProt-node-lookup.db
 # MUSPROT_SUMMARY_PATH=/data/musprot_summary.json
 # MUSPROT_DOCS_PATH=/data/MuSProt_documentation.md
 # MUSPROT_PLOTS_DIR=/data/plots

README.md

@@ -37,6 +37,7 @@ Recommended Dataset repository files:
 
 ```text
 MuSProt.db
+MuSProt-node-lookup.db
 musprot_summary.json
 MuSProt_documentation.md
 plots/
@@ -46,6 +47,7 @@ Optional Space variables:
 
 ```text
 MUSPROT_DB_PATH=/data/MuSProt.db
+MUSPROT_NODE_DB_PATH=/data/MuSProt-node-lookup.db
 MUSPROT_SUMMARY_PATH=/data/musprot_summary.json
 MUSPROT_DOCS_PATH=/data/MuSProt_documentation.md
 MUSPROT_PLOTS_DIR=/data/plots

backend/app/protein/config.py

@@ -66,6 +66,20 @@ def get_summary_path() -> Path | None:
     )
 
 
+def get_node_database_path() -> Path:
+    """Return an optional indexed node lookup sidecar, falling back to the main DB."""
+    explicit_path = os.getenv("MUSPROT_NODE_DB_PATH")
+    if explicit_path:
+        path = Path(explicit_path).expanduser().resolve()
+        if path.exists():
+            return path
+
+    return (
+        _existing_path(DEFAULT_VOLUME_DIR / "MuSProt-node-lookup.db")
+        or get_database_path()
+    ).resolve()
+
+
 def get_docs_path() -> Path | None:
     """Return the dataset documentation path when available."""
     explicit_path = os.getenv("MUSPROT_DOCS_PATH")

backend/app/protein/tsv_loader.py

@@ -8,10 +8,11 @@ key the first occurrence is kept.
 import ast
 import sqlite3
 import logging
+import zlib
 from functools import lru_cache
 from typing import Dict, List, Optional, Tuple, Any
 
-from app.protein.config import get_database_path
+from app.protein.config import get_database_path, get_node_database_path
 from app.protein.database import connect_readonly
 
 logger = logging.getLogger(__name__)
@@ -81,7 +82,7 @@ def reset_index() -> None:
 @lru_cache(maxsize=4096)
 def _lookup_node_row(pdb_id: str, auth_asym_id: str) -> Optional[_Row]:
     """Read one node row without blocking on a full-table in-memory index."""
-    conn = connect_readonly(get_database_path())
+    conn = connect_readonly(get_node_database_path())
     try:
         conn.row_factory = sqlite3.Row
         row = conn.execute(
@@ -93,7 +94,10 @@ def _lookup_node_row(pdb_id: str, auth_asym_id: str) -> Optional[_Row]:
         ).fetchone()
         if row is None:
             return None
-        return {key: row[key] or "" for key in row.keys()}
+        return {
+            key: zlib.decompress(value).decode("utf-8") if isinstance(value, bytes) else value or ""
+            for key, value in ((key, row[key]) for key in row.keys())
+        }
     finally:
         conn.close()
 

backend/scripts/generate_node_lookup.py

@@ -0,0 +1,66 @@
+"""Generate an indexed read-only node lookup sidecar from MuSProt.db."""
+from __future__ import annotations
+
+import argparse
+import sqlite3
+import zlib
+from pathlib import Path
+
+
+NODE_COLUMNS = (
+    "pdb_id", "auth_asym_id", "base_label", "sequence", "sequence_length",
+    "CATH_ID", "cath_superfamily", "Rosetta", "FoldX", "EvoEF2", "RM", "RM+",
+    "ranked_functions", "state_id", "experimental_method", "pH", "temp_K",
+)
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("source", type=Path)
+    parser.add_argument("-o", "--output", type=Path, default=Path("MuSProt-node-lookup.db"))
+    args = parser.parse_args()
+
+    output = args.output.resolve()
+    output.unlink(missing_ok=True)
+
+    conn = sqlite3.connect(output)
+    try:
+        conn.execute("PRAGMA journal_mode = OFF")
+        conn.execute("PRAGMA synchronous = OFF")
+        conn.execute(
+            'CREATE TABLE node (pdb_id, auth_asym_id, base_label, sequence BLOB,'
+            ' sequence_length, CATH_ID, cath_superfamily, Rosetta, FoldX, EvoEF2,'
+            ' RM, "RM+", ranked_functions BLOB, state_id, experimental_method, pH, temp_K)'
+        )
+        source = sqlite3.connect(f"file:{args.source.resolve()}?mode=ro", uri=True)
+        try:
+            select_columns = ", ".join(f'"{column}"' for column in NODE_COLUMNS)
+            rows = source.execute(f"SELECT {select_columns} FROM node")
+            placeholders = ", ".join("?" for _ in NODE_COLUMNS)
+            batch = []
+            for row in rows:
+                row = list(row)
+                for index in (3, 12):
+                    row[index] = zlib.compress((row[index] or "").encode("utf-8"), 1)
+                batch.append(row)
+                if len(batch) == 1000:
+                    conn.executemany(f"INSERT INTO node VALUES ({placeholders})", batch)
+                    batch.clear()
+            if batch:
+                conn.executemany(f"INSERT INTO node VALUES ({placeholders})", batch)
+        finally:
+            source.close()
+        conn.execute(
+            "CREATE INDEX idx_node_chain "
+            "ON node(LOWER(pdb_id), UPPER(auth_asym_id))"
+        )
+        conn.execute("ANALYZE")
+        conn.commit()
+    finally:
+        conn.close()
+
+    print(f"Created {output} ({output.stat().st_size:,} bytes)")
+
+
+if __name__ == "__main__":
+    main()