fix: chembl_bridge()/dti_pairs() conflict under concurrent calls

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "scigantic-bindingdb"
-version = "0.2.2"
+version = "0.2.3"
 description = "Query BindingDB directly from a public S3 mirror with DuckDB, including a ChEMBL cross-reference bridge table and a ready drug-target-interaction training table."
 readme = "README.md"
 requires-python = ">=3.10"

src/scigantic_bindingdb/chembl_bridge.py

@@ -49,10 +49,6 @@ def chembl_bridge(
     con = connect(release)
     try:
         bridge_path = _resolve(f"{release}/derived/bindingdb_chembl_bridge.parquet")
-        con.execute(
-            "CREATE OR REPLACE VIEW chembl_bridge AS "
-            f"SELECT * FROM read_parquet('{bridge_path}')"
-        )
 
         where: list[str] = []
         params: list[str | int] = []
@@ -61,12 +57,22 @@ def chembl_bridge(
             params.append(reactant_set_id)
         clause = f"WHERE {' AND '.join(where)}" if where else ""
 
+        # read_parquet(bridge_path) inline rather than a named
+        # CREATE VIEW: connect() now hands out a cursor on a base
+        # connection shared across calls (see connection.py), and a
+        # named view is catalog state on that shared connection.
+        # Concurrent calls each doing CREATE OR REPLACE VIEW on the same
+        # name raced each other's DDL transaction: reproduced directly,
+        # concurrent chembl_bridge()/dti_pairs() calls from a thread pool
+        # mostly failed with DuckDB's "Catalog write-write conflict".
+        # Referencing the parquet file directly in the query has no
+        # catalog side effect at all, so there's nothing to conflict on.
         if with_names:
             mol_dict = f"s3://{CHEMBL_BUCKET}/{chembl_release}/parquet/molecule_dictionary.parquet"
             sql = f"""
                 SELECT b.reactant_set_id, b.chembl_molregno, b.chembl_id, b.match_method,
                        m.ligand_smiles, d.pref_name AS chembl_pref_name
-                FROM chembl_bridge b
+                FROM read_parquet('{bridge_path}') b
                 JOIN measurements m ON m.reactant_set_id = b.reactant_set_id
                 LEFT JOIN read_parquet('{mol_dict}') d ON d.molregno = b.chembl_molregno
                 {clause}
@@ -75,7 +81,7 @@ def chembl_bridge(
             sql = f"""
                 SELECT b.reactant_set_id, b.chembl_molregno, b.chembl_id, b.match_method,
                        m.ligand_smiles
-                FROM chembl_bridge b
+                FROM read_parquet('{bridge_path}') b
                 JOIN measurements m ON m.reactant_set_id = b.reactant_set_id
                 {clause}
             """

src/scigantic_bindingdb/dti_pairs.py

@@ -54,7 +54,6 @@ def dti_pairs(
     con = connect(release)
     try:
         path = _resolve(f"{release}/derived/dti_pairs.parquet")
-        con.execute(f"CREATE OR REPLACE VIEW dti_pairs AS SELECT * FROM read_parquet('{path}')")
 
         where: list[str] = []
         params: list[str | int] = []
@@ -68,7 +67,13 @@ def dti_pairs(
             where.append("n_chains_declared = 1")
         clause = f"WHERE {' AND '.join(where)}" if where else ""
 
-        sql = f"SELECT * FROM dti_pairs {clause} ORDER BY p_affinity DESC"
+        # read_parquet(path) inline rather than a named CREATE VIEW: see
+        # chembl_bridge.py's comment on the same change. connect() now
+        # returns a cursor on a base connection shared across calls, and a
+        # named view is catalog state on that shared connection --
+        # concurrent calls racing CREATE OR REPLACE VIEW on the same name
+        # hit DuckDB's "Catalog write-write conflict", reproduced directly.
+        sql = f"SELECT * FROM read_parquet('{path}') {clause} ORDER BY p_affinity DESC"
         if limit is not None:
             sql += " LIMIT ?"
             params.append(int(limit))

tests/test_chembl_bridge.py

@@ -1,3 +1,5 @@
+from concurrent.futures import ThreadPoolExecutor
+
 import scigantic_bindingdb as bindingdb
 
 # A gefitinib measurement, verified against the live mirror to resolve
@@ -29,3 +31,25 @@ def test_bridge_total_row_count():
         "'s3://scigantic-bindingdb/202608/derived/bindingdb_chembl_bridge.parquet')"
     )
     assert total["n"].iloc[0] == 2272063
+
+
+def test_concurrent_calls_do_not_conflict_on_catalog():
+    # Regression test for a real bug introduced by connect()'s move to a
+    # shared base connection (see connection.py): chembl_bridge() used to
+    # register its result as a named CREATE OR REPLACE VIEW, which is
+    # catalog-mutating DDL. Two calls racing that DDL on the same shared
+    # connection from different threads mostly failed with DuckDB's
+    # "Catalog write-write conflict", reproduced directly (46/48 calls
+    # failed in a 24-thread stress run) before this was fixed by
+    # referencing the parquet file inline instead of via a named view.
+    def call(i):
+        return bindingdb.chembl_bridge(
+            reactant_set_id=_GEFITINIB_REACTANT_SET_ID, with_names=(i % 2 == 0)
+        )
+
+    with ThreadPoolExecutor(max_workers=16) as pool:
+        results = list(pool.map(call, range(16)))
+
+    for df in results:
+        assert len(df) == 1
+        assert df["chembl_id"].iloc[0] == "CHEMBL939"

tests/test_dti_pairs.py

@@ -1,4 +1,5 @@
 import math
+from concurrent.futures import ThreadPoolExecutor
 
 import scigantic_bindingdb as bindingdb
 
@@ -69,3 +70,26 @@ def test_uniprot_filter():
 def test_sorted_most_potent_first():
     df = bindingdb.dti_pairs(endpoint="ki", uniprot_id="P00533")
     assert df["p_affinity"].is_monotonic_decreasing
+
+
+def test_concurrent_calls_do_not_conflict_on_catalog():
+    # Regression test for a real bug introduced by connect()'s move to a
+    # shared base connection (see connection.py): dti_pairs() used to
+    # register its result as a named CREATE OR REPLACE VIEW, which is
+    # catalog-mutating DDL. Two calls racing that DDL on the same shared
+    # connection from different threads mostly failed with DuckDB's
+    # "Catalog write-write conflict" (reproduced directly before this was
+    # fixed by referencing the parquet file inline instead of via a named
+    # view; see chembl_bridge.py's equivalent test).
+    endpoints = ("ki", "ic50", "kd", "ec50")
+
+    def call(i):
+        endpoint = endpoints[i % len(endpoints)]
+        return endpoint, bindingdb.dti_pairs(endpoint=endpoint, limit=50)
+
+    with ThreadPoolExecutor(max_workers=16) as pool:
+        results = list(pool.map(call, range(16)))
+
+    for endpoint, df in results:
+        assert len(df) == 50
+        assert (df["endpoint"] == endpoint).all()