Patch delays

requirements.txt

@@ -5,7 +5,7 @@ git+https://github.com/falkordb/falkordb-bulk-loader.git@v1.0.6
 setuptools>=66
 pytest
 PyYAML
-git+https://github.com/helxplatform/dug@develop
+git+https://github.com/helxplatform/dug@tranql-crawl-concurrency
 orjson>=3.11
 git+https://github.com/helxplatform/kg_utils.git@v0.0.10.1
 # kg_utils/merging.py hashes an f-string; xxhash 4.0 dropped the

scripts/gzip_artifacts.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python
+"""Gzip-compress annotate/crawl artifacts committed before storage.py's
+write_object started gzipping them (see roger.core.storage).
+
+A dataset annotated before that fix has its elements.txt/concepts.txt sitting
+in lakefs uncompressed -- ~3MB/dir raw for a typical dbGaP data dict pair.
+crawl's own output is gzipped going forward, but crawl still has to
+*download* that old, uncompressed annotation output as input first: for
+bdc-parent's 61,597 dirs that is ~186GB of local disk before crawl writes a
+single byte, which is most of what blew the crawl task's 200G PVC.
+
+Unlike migrate_pickled_classes.py this does no jsonpickle decode/re-encode --
+pure byte-level gzip, so the serialized content is untouched, just smaller.
+
+    lakectl local clone lakefs://<repo>/<branch>/annotate_and_index/<task>/ ./out
+    python scripts/gzip_artifacts.py ./out            # compress in place
+    lakectl local commit ./out -m "gzip annotation output"
+
+    python scripts/gzip_artifacts.py --dry-run ./out  # report only
+    python scripts/gzip_artifacts.py --self-check     # no dir needed
+"""
+
+import argparse
+import gzip
+import sys
+from pathlib import Path
+
+ARTIFACTS = ('elements.txt', 'concepts.txt', 'expanded_concepts.txt')
+GZIP_MAGIC = b'\x1f\x8b'
+# level 6 matches storage.py's write_object -- level 9 (gzip's default)
+# burned 3-4x the cpu for the same ratio on this repetitive JSON.
+COMPRESSLEVEL = 6
+
+
+def artifact_files(root):
+    return sorted(p for p in Path(root).rglob('*.txt') if p.name in ARTIFACTS)
+
+
+def compress_file(path, dry_run=False):
+    """Returns (raw_size, compressed_size) if compressed, None if skipped
+    (already gzip or empty)."""
+    raw = path.read_bytes()
+    if not raw or raw[:2] == GZIP_MAGIC:
+        return None
+    compressed = gzip.compress(raw, compresslevel=COMPRESSLEVEL)
+    if not dry_run:
+        tmp = path.with_name(path.name + '.gzip-tmp')
+        tmp.write_bytes(compressed)
+        tmp.replace(path)
+    return len(raw), len(compressed)
+
+
+def run(root, dry_run=False):
+    files = artifact_files(root)
+    total_raw = total_compressed = 0
+    changed = skipped = 0
+    for i, path in enumerate(files, 1):
+        result = compress_file(path, dry_run=dry_run)
+        if result is None:
+            skipped += 1
+            continue
+        raw_size, compressed_size = result
+        total_raw += raw_size
+        total_compressed += compressed_size
+        changed += 1
+        if changed % 5000 == 0:
+            print(f"  {changed} compressed ({i}/{len(files)} scanned)")
+    verb = "would compress" if dry_run else "compressed"
+    print(f"{verb} {changed} of {len(files)} file(s), "
+          f"{skipped} already gzip or empty")
+    if total_raw:
+        print(f"{total_raw / 1e9:.2f}GB -> {total_compressed / 1e9:.2f}GB "
+              f"({total_raw / total_compressed:.1f}x)")
+    return changed
+
+
+def self_check():
+    "Round-trip a small payload; fails loudly if compress_file regresses."
+    import tempfile
+    payload = b'{"id": "UMLS:C1"}' * 500
+
+    with tempfile.TemporaryDirectory() as tmp:
+        path = Path(tmp) / 'concepts.txt'
+        path.write_bytes(payload)
+
+        result = compress_file(path, dry_run=True)
+        assert result is not None, "dry-run should report a would-be change"
+        assert path.read_bytes() == payload, "dry-run must not touch the file"
+
+        result = compress_file(path)
+        assert result is not None
+        raw_size, compressed_size = result
+        assert raw_size == len(payload)
+        assert compressed_size < raw_size, "should have actually shrunk"
+
+        with open(path, 'rb') as f:
+            assert f.read(2) == GZIP_MAGIC
+        assert gzip.decompress(path.read_bytes()) == payload
+
+        # idempotent: running again on an already-gzipped file is a no-op
+        assert compress_file(path) is None
+        assert gzip.decompress(path.read_bytes()) == payload
+
+    print("self-check ok")
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__,
+                                     formatter_class=argparse.RawDescriptionHelpFormatter)
+    parser.add_argument('root', nargs='?', help='directory to compress in place')
+    parser.add_argument('--dry-run', action='store_true',
+                        help='report what would change without writing')
+    parser.add_argument('--self-check', action='store_true',
+                        help='round-trip a synthetic payload, no dir needed')
+    args = parser.parse_args()
+
+    if args.self_check:
+        self_check()
+        return
+    if not args.root:
+        parser.error("root directory required unless --self-check")
+    run(args.root, dry_run=args.dry_run)
+
+
+if __name__ == '__main__':
+    sys.exit(main())

scripts/migrate_pickled_classes.py

@@ -16,6 +16,7 @@
 """
 
 import argparse
+import gzip
 import importlib
 import os
 import json
@@ -39,6 +40,23 @@
 PY_OBJECT = re.compile(r'"py/object":\s*"([^"]+)"')
 ARTIFACTS = ('elements.txt', 'concepts.txt', 'expanded_concepts.txt')
 
+GZIP_MAGIC = b'\x1f\x8b'
+
+
+def read_artifact_text(path):
+    """roger's storage.write_object gzips these now; older artifacts
+    committed before that are still plain text, so detect and handle both."""
+    raw = path.read_bytes()
+    if raw[:2] == GZIP_MAGIC:
+        return gzip.decompress(raw).decode('utf-8')
+    return raw.decode('utf-8')
+
+
+def write_artifact_text(path, text):
+    """Always write gzip -- migration is also the chance to bring an old
+    plain-text artifact up to the current on-disk format."""
+    path.write_bytes(gzip.compress(text.encode('utf-8')))
+
 
 def install_alias(module_path):
     """Stand in for a legacy module, resolving class names against CURRENT."""
@@ -110,7 +128,7 @@ def field_drift_paths(paths):
     """
     drift = {}
     for path in paths:
-        obj = jsonpickle.decode(path.read_text())
+        obj = jsonpickle.decode(read_artifact_text(path))
         stack, seen = [obj], set()
         while stack:
             item = stack.pop()
@@ -137,7 +155,7 @@ def scan(root):
     print(f"{len(files)} artifact file(s) under {root}")
     found = set()
     for path in files:
-        found |= classes_in(path.read_text())
+        found |= classes_in(read_artifact_text(path))
     broken = broken_modules(found)
     for cls in sorted(found):
         module_path = cls.rpartition('.')[0]
@@ -180,7 +198,7 @@ def dead_modules(sample_paths):
     """
     dead, found = set(), set()
     for path in sample_paths:
-        found |= classes_in(path.read_text())
+        found |= classes_in(read_artifact_text(path))
     for cls in found:
         module_path = cls.rpartition('.')[0]
         if module_path in dead or module_path in sys.modules:
@@ -271,7 +289,7 @@ def restamp(root, sample=20, dry_run=False):
 
     changed = 0
     for i, path in enumerate(files, 1):
-        text = path.read_text()
+        text = read_artifact_text(path)
         new_text, unmapped = restamp_text(text, dead)
         if unmapped:
             raise SystemExit(
@@ -284,7 +302,7 @@ def restamp(root, sample=20, dry_run=False):
             # write-then-rename: a pod killed mid-write must not leave a
             # truncated artifact behind, and there are 150k of them
             tmp = path.with_name(path.name + '.restamp-tmp')
-            tmp.write_text(new_text)
+            write_artifact_text(tmp, new_text)
             os.replace(tmp, path)
         if changed % 5000 == 0:
             print(f"  {changed} rewritten ({i}/{len(files)} scanned)")
@@ -296,13 +314,13 @@ def restamp(root, sample=20, dry_run=False):
 def fix(root, dry_run=False):
     files = artifact_files(root)
     for module_path in broken_modules(
-            {c for p in files for c in classes_in(p.read_text())}):
+            {c for p in files for c in classes_in(read_artifact_text(p))}):
         print(f"aliasing legacy module {module_path}")
         install_alias(module_path)
 
     changed = 0
     for path in files:
-        text = path.read_text()
+        text = read_artifact_text(path)
         obj = jsonpickle.decode(text)
         fill_defaults(obj)
         rewritten = jsonpickle.encode(obj, indent=2)
@@ -311,7 +329,7 @@ def fix(root, dry_run=False):
         changed += 1
         print(f"{'would rewrite' if dry_run else 'rewrote'} {path}")
         if not dry_run:
-            path.write_text(rewritten)
+            write_artifact_text(path, rewritten)
     print(f"{changed} of {len(files)} file(s) needed migration")
     return changed
 

src/roger/config/__init__.py

@@ -232,6 +232,19 @@ class IndexingConfig(DictLike):
         "anat_to_pheno": ["anatomical_entity", "phenotypic_feature"],
     })
     tranql_endpoint: str = "http://tranql-service/tranql/query?dynamic_id_resolution=true&asynchronous=false"
+    # Concurrency for the crawl. `crawl_workers` threads the TranQL fetches
+    # within one concept; `crawl_file_workers` threads whole input files.
+    # They multiply, and the product should not exceed what the TranQL
+    # service can serve at once (its gunicorn worker count) -- past that,
+    # requests only queue.
+    crawl_workers: int = 4
+    crawl_file_workers: int = 4
+    # crawl_file_workers threads doing real work (TranQL fetches, jsonpickle
+    # encode, gzip) on the chart's default cpu limit throttled a crawl pod
+    # 70% of its scheduling periods, cutting throughput to a third. Match
+    # crawl_file_workers 1:1 with cores so each worker thread gets its own,
+    # instead of 4 threads fighting over a fraction of one.
+    crawl_cpu: str = "4"
     # by default skips node to element queries
     node_to_element_queries: dict = field(default_factory=lambda: {})
     element_mapping: str = ""

src/roger/config/config.yaml

@@ -120,6 +120,9 @@ indexing:
     "chemical_mixture_to_disease": ["chemical_mixture", "disease"]
     "phen_to_anat": ["phenotypic_feature", "anatomical_entity"]
   tranql_endpoint: "http://tranql-service/tranql/query?dynamic_id_resolution=true&asynchronous=false"
+  crawl_workers: 4
+  crawl_file_workers: 4
+  crawl_cpu: "4"
   node_to_element_queries:
     enabled: false
     cde:

src/roger/core/bulkload.py

@@ -54,12 +54,19 @@ def create_nodes_csv_file(self, input_data_path=None, output_data_path=None):
         merged_nodes_file = storage.merged_objects('nodes', input_data_path)
         counter = 1
         for node in storage.json_line_iter(merged_nodes_file):
+            # \r alone (no paired \n) still reads as a line break under
+            # universal newlines -- both our own row count and the bulk
+            # loader's CSV parser split on it, turning one row into two
+            # and corrupting the column count. Seen in dbGaP codebook text
+            # pasted in with stray CRs (e.g. a Study "activitybk" field).
             if node.get('description'):
-                node['description'] = node['description'].replace('\n',
-                                                                  ' ')
+                node['description'] = (
+                    node['description'].replace('\r\n', ' ')
+                    .replace('\r', ' ').replace('\n', ' '))
             if node.get('name'):
-                node['name'] = node['name'].replace('\n',
-                                                    ' ')
+                node['name'] = (
+                    node['name'].replace('\r\n', ' ')
+                    .replace('\r', ' ').replace('\n', ' '))
             if not node.get('category'):
                 category_error_nodes.add(node['id'])
                 node['category'] = [BiolinkModel.root_type]

src/roger/core/storage.py

@@ -5,6 +5,7 @@
 
 import os
 import glob
+import gzip
 import time
 import pathlib
 import pickle
@@ -86,9 +87,25 @@ def read_object(path, key=None):
         with open(file=path, mode="rb") as stream:
             obj = pickle.load(stream)
     elif path.endswith(".jsonl") or path.endswith('.txt'):
-        obj = read_data(path)
+        obj = read_gzip_or_plain_text(path) if not is_web(path) \
+            else read_data(path)
     return obj
 
+# gzip magic number -- distinguishes a compressed artifact from the plain
+# text ones already committed to lakefs before this was added, so both
+# read transparently and nothing downstream needs to change.
+GZIP_MAGIC = b'\x1f\x8b'
+
+def read_gzip_or_plain_text(path):
+    """ Read a local .txt/.jsonl artifact, decompressing it if it was
+    gzip-written by write_object; falls back to plain text for artifacts
+    written before compression was added. """
+    with open(path, 'rb') as stream:
+        raw = stream.read()
+    if raw[:2] == GZIP_MAGIC:
+        return gzip.decompress(raw).decode('utf-8')
+    return raw.decode('utf-8')
+
 def is_web (uri):
     """ The URI is a web URI (starts with http or https).
     :param uri: A URI """
@@ -122,8 +139,19 @@ def write_object (obj, path, key=None):
         with open (path, "wb") as stream:
             pickle.dump(obj, file=stream)
     elif path.endswith(".jsonl") or path.endswith('.txt'):
-        with open (path, "w", encoding="utf-8") as stream:
-            stream.write(obj)
+        # gzip -- these are the crawl-stage KG-answer artifacts, and the
+        # same repeated CURIEs/biolink categories/JSON keys compress
+        # 5-10x; that's the difference between fitting a large dataset's
+        # crawl output on the PVC and hitting ENOSPC mid-run. Same
+        # filename/extension as before so every glob pattern that finds
+        # these files by name still matches; read_object detects the
+        # gzip magic number so old uncompressed artifacts still read.
+        with open(path, "wb") as stream:
+            # level 9 (gzip.compress's default) burned 3-4x the CPU of
+            # level 6 for the same ratio on this repetitive JSON -- deadly
+            # under a crawl task's thin CPU limit (measured 70% of periods
+            # throttled on a 250m limit with 4 crawl workers).
+            stream.write(gzip.compress(obj.encode('utf-8'), compresslevel=6))
     else:
         # Raise an exception if invalid.
         raise ValueError (f"Unrecognized extension: {path}")