Pathfinder to use Retriever instead of Gandalf

.github/ISSUE_TEMPLATE/tier0-arax-rollout.md

@@ -13,7 +13,6 @@ assignees: ""
 - [ ] Build database file `tier0-info-for-overlay_v1.0_tier0-MMDDYYYY.sqlite` (assignee: ; subissue: ) (build script: [`generate_sqlite.py`](https://github.com/RTXteam/RTX/blob/master/code/ARAX/KnowledgeSources/generate_sqlite.py))
 - [ ] Build database file `curie_ngd_v1.0_tier0-MMDDYYYY.sqlite` (assignee: ; subissue: )
 - [ ] Build database file `ExplainableDTD_v1.0_tier0-MMDDYYYY-all_with_paths.db` (assignee: ; subissue: )
-- [ ] gandalf_mmap tarball refresh work. Work with PSU team on this. (assignee: ; subissue: )
 - [ ] Update ARAX `config_dbs.json` for the new database files. (assignee: ; subissue: )
 - [ ] Update `ARAX_database_manager.py` for the new database files. (assignee: ; subissue: )
 - [ ] Stage all rebuilt Tier0 KG-based ARAX database files on the following servers: (assignee: ; subissue: )

.gitignore

@@ -53,8 +53,6 @@ code/ARAX/KnowledgeSources/meta_kg*.json
 code/ARAX/KnowledgeSources/Prediction/*.pkl
 code/ARAX/KnowledgeSources/Prediction/*.txt
 code/ARAX/KnowledgeSources/Prediction/*.gz
-code/ARAX/KnowledgeSources/Gandalf/*.gz
-code/ARAX/KnowledgeSources/Gandalf/gandalf_mmap
 
 code/ARAX/NodeSynonymizer/*.tsv
 code/ARAX/NodeSynonymizer/*.pickle

code/ARAX/ARAXQuery/ARAX_connect.py

@@ -1,6 +1,7 @@
 import json
 import sys
 
+import requests
 from RTXConfiguration import RTXConfiguration
 from pathfinder.Pathfinder import Pathfinder
 from xcrg import XCRGConfig, run_xcrg
@@ -15,7 +16,7 @@ def eprint(*args, **kwargs): print(*args, file=sys.stderr, **kwargs)
 import time
 
 sys.path.append(os.path.dirname(os.path.abspath(__file__)))
-from Path_Finder.utility import get_curie_ngd_path, get_curie_to_pmids_path, get_kg2c_db_path, get_gandalf_mmap_path
+from Path_Finder.utility import get_curie_ngd_path, get_curie_to_pmids_path, get_kg2c_db_path
 from Expand.trapi_query_cacher import KPQueryCacher
 from ARAX_messenger import ARAXMessenger
 
@@ -462,10 +463,10 @@ def __connect_nodes(self, describe=False):
         descendants = []
         if category_constraint:
             descendants = set(get_biolink_helper().get_descendants(category_constraint))
-
+        rtx_config = RTXConfiguration()
+        retriever_url = get_xcrg_retriever_url(rtx_config)
         pathfinder = Pathfinder(
-            "MLRepo",
-            get_gandalf_mmap_path(),
+            f"retriever:{retriever_url}",
             get_curie_ngd_path(),
             get_kg2c_db_path(),
             blocked_curies,
@@ -498,7 +499,7 @@ def __connect_nodes(self, describe=False):
                                   f"with a max path length of {self.parameters['max_path_length']}.")
             return self.response
 
-        self.convert_to_trapi(result, aux_graphs, knowledge_graph)
+        self.convert_to_trapi(result, aux_graphs, knowledge_graph, retriever_url)
 
         mode = 'ARAX'
         if mode != "RTXKG2" and not hasattr(self.response, "original_query_graph"):
@@ -586,7 +587,7 @@ def get_block_list(self):
         synonyms = set(s.lower() for s in json_block_list['synonyms'])
         return set(json_block_list['curies']), synonyms
 
-    def convert_to_trapi(self, result, aux_graphs, knowledge_graph):
+    def convert_to_trapi(self, result, aux_graphs, knowledge_graph, retriever_url):
         if self.response.envelope.message.results is None:
             self.response.envelope.message.results = []
 
@@ -595,8 +596,8 @@ def convert_to_trapi(self, result, aux_graphs, knowledge_graph):
 
         if self.response.envelope.message.knowledge_graph is None:
             self.response.envelope.message.knowledge_graph = {}
-
-        kg = KnowledgeGraph().from_dict(knowledge_graph)
+        rehydrated_kg = self.rehydrate(knowledge_graph, retriever_url)
+        kg = KnowledgeGraph().from_dict(rehydrated_kg)
 
         analyses = []
         for analys in result['analyses']:
@@ -630,6 +631,47 @@ def convert_to_trapi(self, result, aux_graphs, knowledge_graph):
         self.response.envelope.message.knowledge_graph.edges.update(kg.edges)
         self.response.envelope.message.knowledge_graph.nodes.update(kg.nodes)
 
+    def rehydrate(self, kg, retriever_url):
+        headers = {"Content-Type": "application/json", "Accept": "application/json"}
+        payload = {
+            "message": {
+                "knowledge_graph": kg
+            },
+            "parameters": {
+                "rehydrate": True,
+                "tier": 0
+            }
+        }
+
+        try:
+            res = requests.post(
+                retriever_url.replace("query", "rehydrate"), headers=headers, json=payload, timeout=30
+            )
+            res.raise_for_status()
+            return res.json()["message"]["knowledge_graph"]
+
+        except requests.exceptions.HTTPError as http_err:
+            self.response.error(f"HTTP error occurred: {http_err}")
+            if res.text:
+                self.response.error(f"Error details: {res.text}")
+            raise http_err
+        except requests.exceptions.ConnectionError as conn_err:
+            self.response.error(f"Connection error occurred: {conn_err}")
+            raise conn_err
+        except requests.exceptions.Timeout as timeout_err:
+            self.response.error(f"Timeout error occurred: {timeout_err}")
+            raise timeout_err
+        except requests.exceptions.RequestException as req_err:
+            self.response.error(f"An unexpected error occurred: {req_err}")
+            raise req_err
+        except json.JSONDecodeError:
+            self.response.error("Failed to parse the response as JSON.")
+            self.response.error(f"Raw response: {res.text}")
+            raise json.JSONDecodeError
+        except Exception as e:
+            self.response.error(f"An unexpected error occurred: {e}")
+            raise e
+
 
 ##########################################################################################
 def main():

code/ARAX/ARAXQuery/ARAX_database_manager.py

@@ -54,10 +54,6 @@ def __init__(self, allow_downloads=False):
         ngd_filepath = os.path.sep.join([*pathlist[:(RTXindex + 1)], 'code', 'ARAX', 'KnowledgeSources', 'NormalizedGoogleDistance'])
         if not  os.path.exists(ngd_filepath):
             _run_cmd_in_shell_chk_status(f"mkdir -p {shlex.quote(ngd_filepath)}")
-
-        gandalf_mmap_filepath = os.path.sep.join([*pathlist[:(RTXindex + 1)], 'code', 'ARAX', 'KnowledgeSources', 'Gandalf'])
-        if not  os.path.exists(gandalf_mmap_filepath):
-            _run_cmd_in_shell_chk_status(f"mkdir -p {shlex.quote(gandalf_mmap_filepath)}")
 
         cohd_filepath = os.path.sep.join([*pathlist[:(RTXindex + 1)], 'code', 'ARAX', 'KnowledgeSources', 'COHD_local', 'data'])
         if not  os.path.exists(cohd_filepath):
@@ -83,7 +79,6 @@ def __init__(self, allow_downloads=False):
             'cohd_database': f"{cohd_filepath}{os.path.sep}{self.RTXConfig.cohd_database_path.split('/')[-1]}",
             'curie_to_pmids': f"{ngd_filepath}{os.path.sep}{self.RTXConfig.curie_to_pmids_path.split('/')[-1]}",
             'curie_ngd': f"{ngd_filepath}{os.path.sep}{self.RTXConfig.curie_ngd_path.split('/')[-1]}",
-            'gandalf_mmap': f"{gandalf_mmap_filepath}{os.path.sep}{self.RTXConfig.gandalf_mmap_path.split('/')[-1]}",
             'kg2c_sqlite': f"{kg2c_filepath}{os.path.sep}{self.RTXConfig.kg2c_sqlite_path.split('/')[-1]}",
             'fda_approved_drugs': f"{fda_approved_drugs_filepath}{os.path.sep}{self.RTXConfig.fda_approved_drugs_path.split('/')[-1]}",
             'autocomplete': f"{autocomplete_filepath}{os.path.sep}{self.RTXConfig.autocomplete_path.split('/')[-1]}",
@@ -97,7 +92,6 @@ def __init__(self, allow_downloads=False):
             'cohd_database': self.get_database_subpath(self.RTXConfig.cohd_database_path),
             'curie_to_pmids': self.get_database_subpath(self.RTXConfig.curie_to_pmids_path),
             'curie_ngd': self.get_database_subpath(self.RTXConfig.curie_ngd_path),
-            'gandalf_mmap': self.get_database_subpath(self.RTXConfig.gandalf_mmap_path),
             'kg2c_sqlite': self.get_database_subpath(self.RTXConfig.kg2c_sqlite_path),
             'fda_approved_drugs': self.get_database_subpath(self.RTXConfig.fda_approved_drugs_path),
             'autocomplete': self.get_database_subpath(self.RTXConfig.autocomplete_path),
@@ -109,7 +103,6 @@ def __init__(self, allow_downloads=False):
             'cohd_database': self.get_remote_location('cohd_database'),
             'curie_to_pmids': self.get_remote_location('curie_to_pmids'),
             'curie_ngd': self.get_remote_location('curie_ngd'),
-            'gandalf_mmap': self.get_remote_location('gandalf_mmap'),
             'kg2c_sqlite': self.get_remote_location('kg2c_sqlite'),
             'fda_approved_drugs': self.get_remote_location('fda_approved_drugs'),
             'autocomplete': self.get_remote_location('autocomplete'),
@@ -121,7 +114,6 @@ def __init__(self, allow_downloads=False):
             'cohd_database': self.get_docker_path('cohd_database'),
             'curie_to_pmids': self.get_docker_path('curie_to_pmids'),
             'curie_ngd': self.get_docker_path('curie_ngd'),
-            'gandalf_mmap': self.get_docker_path('gandalf_mmap'),
             'kg2c_sqlite': self.get_docker_path('kg2c_sqlite'),
             'fda_approved_drugs': self.get_docker_path('fda_approved_drugs'),
             'autocomplete': self.get_docker_path('autocomplete'),
@@ -142,10 +134,6 @@ def __init__(self, allow_downloads=False):
                 'path': self.local_paths['curie_ngd'],
                 'version': self.RTXConfig.curie_ngd_version
             },
-            'gandalf_mmap': {
-                'path': self.local_paths['gandalf_mmap'],
-                'version': self.RTXConfig.gandalf_mmap_version
-            },
             'kg2c_sqlite': {
                 'path': self.local_paths['kg2c_sqlite'],
                 'version': self.RTXConfig.kg2c_sqlite_version
@@ -224,8 +212,7 @@ def update_databases(self, debug = True, response = None):
                                             local_symlink_target_path=self.docker_central_paths[database_name],
                                             debug=debug)
                 # tarball is present but the unpacked dir next to the real archive
-                # is missing; re-extract in place without re-running rsync (avoids
-                # re-downloading large tarballs like gandalf_mmap on every restart).
+                # is missing; re-extract in place without re-running rsync
                 # realpath follows the symlink to the docker-central side so the
                 # check matches where _extract_tarball actually puts the unpacked dir.
                 elif local_path.endswith('.tar.gz') and not os.path.isdir(
@@ -370,7 +357,7 @@ def _download_database(self, remote_location, local_destination_path, local_syml
 
     def _extract_tarball(self, tarball_path, debug=False):
         # follow the symlink to the real archive so extraction lands next to the
-        # docker-central tarball (where Pathfinder reads it via get_gandalf_mmap_path),
+        # docker-central tarball,
         # not next to the RTX-side symlink. realpath is a no-op on dev machines
         # where the path is already a real file.
         resolved = os.path.realpath(tarball_path)

code/ARAX/ARAXQuery/Path_Finder/utility.py

@@ -24,38 +24,10 @@ def get_curie_to_pmids_path():
     sqlite_name = RTXConfiguration().curie_to_pmids_path.split("/")[-1]
     return f"sqlite:{filepath}{os.path.sep}{sqlite_name}"
 
-def get_gandalf_mmap_path():
-    # The ARAX_database_manager.py module unpacks the tarball
-    # /mnt/data/orangeboard/databases/tier0-20260408/gandalf_mmap_tier0-20260408.tar.gz"
-    # into files underneath a new subdirectory "gandalf_mmap", like this:
-    # /mnt/data/orangeboard/databases/tier0-20260408/gandalf_mmap/..."
-    # and the database manager creates a symbolic link:
-    # RTX/code/ARAX/KnowledgeSources/Gandalf/gandalf_mmap_tier0-20260408.tar.gz => \
-    #     /mnt/data/orangeboard/databases/tier0-20260408/gandalf_mmap_tier0-20260408.tar.gz
-    # This function is supposed to return "gandalf:/mnt/data/orangeboard/databases/tier0-20260408/gandalf_mmap"
-    # but how can it construct that path?  It has to do the following:
-    # (1) find the local "RTX/code/ARAX/KnowledgeSources/Gandalf" directory
-    # (2) get the symbolic link "gandalf_mmap_tier0-20260408.tar.gz" from that directory
-    #     (where the specific filename comes from RTXConfiguration().gandalf_mmap_path)
-    # (3) resolve that symlink to absolute path
-    #     /mnt/data/orangeboard/databases/tier0-20260408/gandalf_mmap_tier0-20260408.tar.gz
-    # (4) remove the filename and append "/gandalf_mmap"
-    gandalf_mmap_path = RTXConfiguration().gandalf_mmap_path
-    gandalf_db_bundle = os.path.basename(gandalf_mmap_path)
-    gandalf_dir = os.path.normpath(os.path.join(os.path.dirname(os.path.realpath(__file__)), '..', '..', 'KnowledgeSources', 'Gandalf'))
-    gandalf_db_bundle_local = os.path.join(gandalf_dir, gandalf_db_bundle)
-    if os.path.islink(gandalf_db_bundle_local):
-        gandalf_db_bundle_to_use = os.path.realpath(gandalf_db_bundle_local)
-    else:
-        gandalf_db_bundle_to_use = gandalf_db_bundle_local
-    gandalf_mmap_dir = os.path.join(os.path.dirname(gandalf_db_bundle_to_use), 'gandalf_mmap')
-    return f"gandalf:{gandalf_mmap_dir}"
-
 
 def main():
     print(get_kg2c_db_path())
     print(get_curie_ngd_path())
-    print(get_gandalf_mmap_path())
 
 
 if __name__ == "__main__":

code/RTXConfiguration.py

@@ -167,8 +167,6 @@ def _private_init(self):
         self.curie_to_pmids_version = self.curie_to_pmids_path.split('/')[-1].split('_v')[-1].replace('.sqlite', '')
         self.curie_ngd_path = database_downloads["curie_ngd"]
         self.curie_ngd_version = self.curie_ngd_path.split('/')[-1].split('_v')[-1].replace('.sqlite', '')
-        self.gandalf_mmap_path = database_downloads["gandalf_mmap"]
-        self.gandalf_mmap_version = self.gandalf_mmap_path.split('/')[-1].split('_v')[-1].replace('.tar.gz', '')
         self.kg2c_sqlite_path = database_downloads["kg2c_sqlite"]
         self.kg2c_sqlite_version = self.kg2c_sqlite_path.split('/')[-1].split('_v')[-1].replace('.sqlite', '')
         self.tier0_sqlite_path = database_downloads["tier0_sqlite"]

code/config_dbs.json

@@ -6,7 +6,6 @@
       "autocomplete": "/translator/data/orangeboard/databases/tier0-20260621/autocomplete_v1.0_tier0-20260621.sqlite",
       "curie_to_pmids": "/translator/data/orangeboard/databases/tier0-20260621/curie_to_pmids_v1.0_tier0-20260621.sqlite",
       "curie_ngd": "/translator/data/orangeboard/databases/tier0-20260621/curie_ngd_v1.0_tier0-20260621.sqlite",
-      "gandalf_mmap": "/translator/data/orangeboard/databases/tier0-20260621/gandalf_mmap_tier0-20260621-gandalf_csr-0_4_2.tar.gz",
       "explainable_dtd_db": "/translator/data/orangeboard/databases/tier0-20260621/ExplainableDTD_v1.0_tier0-20260621-all_with_paths.db",
       "cohd_database": "/translator/data/orangeboard/databases/KG2.8.0/COHDdatabase_v1.0_KG2.8.0.db"
    },

code/generate-db-symlinks.sh

@@ -50,4 +50,3 @@ link "${DB_DIR}/${LATEST_TIER0_VER}/curie_to_pmids_v1.0_${LATEST_TIER0_VER}.sqli
 link "${DB_DIR}/${LATEST_TIER0_VER}/fda_approved_drugs_v1.0.pickle" "RTX/code/ARAX/KnowledgeSources/fda_approved_drugs_v1.0.pickle"
 link "${DB_DIR}/${LATEST_TIER0_VER}/tier0-info-for-overlay_v1.0_${LATEST_TIER0_VER}.sqlite" "RTX/code/ARAX/KnowledgeSources/KG2c/tier0-info-for-overlay_v1.0_${LATEST_TIER0_VER}.sqlite"
 link "${DB_DIR}/${LATEST_TIER0_VER}/autocomplete_v1.0_${LATEST_TIER0_VER}.sqlite" "RTX/code/autocomplete/autocomplete_v1.0_${LATEST_TIER0_VER}.sqlite"
-link "${DB_DIR}/${LATEST_TIER0_VER}/gandalf_mmap_${LATEST_TIER0_VER}-gandalf_csr-0_4_2.tar.gz" "RTX/code/ARAX/KnowledgeSources/Gandalf/gandalf_mmap_${LATEST_TIER0_VER}-gandalf_csr-0_4_2.tar.gz"

deploy/arax/scripts/download_database.sh

@@ -59,11 +59,7 @@ done
 # this is useful when a new db (or a new version of existing db) rolls out
 # and we only want to keep what is specificed in config_dbs.json file
 echo "Removing outdated db files......"
-for existing_db in $(find "$db_folder" \
-                          \( -path "${db_folder}/tier0-*/gandalf_mmap" \
-                             -o -path "${db_folder}/tier0-*/*.tar.gz-unpacked" \) \
-                             -prune -o \
-                     -type f -print)
+for existing_db in $(find "$db_folder" -type f)
 do
   # check if filename ends with .md5, which is NOT db file
   if [[ ! "${existing_db}" == *.md5 ]]

requirements.txt

@@ -60,7 +60,7 @@ opentelemetry-proto==1.17.0
 opentelemetry-sdk==1.17.0
 opentelemetry-semantic-conventions==0.38b0
 opentelemetry-util-http==0.38b0
-catrax-pathfinder==2.3.0
+catrax-pathfinder==2.4.3
 catrax-xcrg @ git+https://github.com/Translator-CATRAX/xCRG.git@c97da5349a432e29d7a6432ee272a8c1311de9da
 biolink-helper-pkg==1.0.1
 httpx