diff --git a/src/test/term_info_queries_test.py b/src/test/term_info_queries_test.py index 8164101..1f16bed 100644 --- a/src/test/term_info_queries_test.py +++ b/src/test/term_info_queries_test.py @@ -416,8 +416,12 @@ def test_term_info_serialization_dataset(self): self.assertFalse("logo" in serialized) self.assertTrue("link" in serialized) self.assertEqual("[http://flybase.org/reports/FBrf0221438.html](http://flybase.org/reports/FBrf0221438.html)", serialized["link"]) - self.assertEqual(4, len(serialized["types"])) - self.assertTrue("DataSet" in serialized["types"]) + # Assert the types this test is about, not how many there are. The + # count moved 4 -> 5 when DataSets gained stage labels ("Adult" here), + # which is backend content: a new label is not a regression, a missing + # DataSet type would be. + for expected_type in ("Entity", "Individual", "DataSet", "has_image"): + self.assertIn(expected_type, serialized["types"]) self.assertEqual("An exhaustive set of lineage clones covering the adult brain from Kei Ito's lab.", serialized["description"]) self.assertFalse("synonyms" in serialized) self.assertFalse("source" in serialized) diff --git a/src/test/test_term_info_fallback.py b/src/test/test_term_info_fallback.py index 6c3b974..4e4d1f1 100644 --- a/src/test/test_term_info_fallback.py +++ b/src/test/test_term_info_fallback.py @@ -128,6 +128,51 @@ def test_schema_version_is_never_a_git_call(monkeypatch, tmp_path): assert tif.schema_version() == "unpinned" +# --------------------------------------------------------------------------- +# Pointing the indexer at VFBquery's own backends +# --------------------------------------------------------------------------- + +@pytest.mark.parametrize("url,expected", [ + ("http://solr.virtualflybrain.org/solr/vfb_json/", + ("http://solr.virtualflybrain.org/solr", "vfb_json")), + ("https://solr.virtualflybrain.org/solr/vfb_json", + ("https://solr.virtualflybrain.org/solr", "vfb_json")), + ("", (None, None)), + (None, (None, None)), +]) +def test_solr_url_splits_into_the_indexers_two_env_vars(url, expected): + assert tif._split_solr_url(url) == expected + + +def test_seeding_sets_a_usable_solr_update_url(monkeypatch): + """The regression this exists for: solr_client builds its update URL from + SOLRserver/SOLRcollection, neither of which a VFBquery deployment sets. + get_solr_update_url() then returns None and send_solr_payload returns + False without making a request, so a rebuilt document is served but never + indexed -- silently, and only in production, because the cache-disabled + guard short-circuits before the URL is ever needed.""" + for key in ("SOLRserver", "SOLRcollection", "PDBserver", "PDBuser", + "PDBpassword"): + monkeypatch.delenv(key, raising=False) + tif._seed_indexer_env() + from vfbquery.vfb_queries import vfb_solr + server, collection = tif._split_solr_url(vfb_solr.url) + assert os.environ["SOLRserver"] == server + assert os.environ["SOLRcollection"] == collection + # and the two together must produce a real update endpoint + assert f"{server}/{collection}/update" == ( + f"{os.environ['SOLRserver'].rstrip('/')}/" + f"{os.environ['SOLRcollection']}/update") + + +def test_seeding_does_not_override_a_deployment(monkeypatch): + monkeypatch.setenv("SOLRserver", "http://elsewhere/solr") + monkeypatch.setenv("SOLRcollection", "other_core") + tif._seed_indexer_env() + assert os.environ["SOLRserver"] == "http://elsewhere/solr" + assert os.environ["SOLRcollection"] == "other_core" + + # --------------------------------------------------------------------------- # The `src` package collision # --------------------------------------------------------------------------- @@ -164,3 +209,127 @@ def test_reports_why_it_cannot_build(monkeypatch): assert tif.fallback_available() is False assert "no indexer here" in tif.fallback_unavailable_reason() assert tif.backfill_term_info("FBbt_00003748") is None + + +# --------------------------------------------------------------------------- +# Per-field query index rebuilds +# --------------------------------------------------------------------------- + +class _StubNeo: + def commit_list(self, queries): + return None + + +class _StubConnect: + """vc.nc is a read-only property, so stand in for vc itself.""" + nc = _StubNeo() + +def test_query_field_fallback_only_claims_fields_it_can_build(monkeypatch): + monkeypatch.setattr(tif, "_QUERY_FIELD_INDEXERS", {"anat_query": object}) + monkeypatch.setattr(tif, "_INDEXERS", {"class": object}) + assert tif.query_field_fallback_available("anat_query") is True + assert tif.query_field_fallback_available("term_info") is False + assert tif.query_field_fallback_available("something_else") is False + + +def test_no_ids_means_no_work(monkeypatch): + def explode(*a, **k): + raise AssertionError("should not have queried anything") + monkeypatch.setattr(tif, "_QUERY_FIELD_INDEXERS", {"anat_query": explode}) + assert tif.backfill_query_field([], "anat_query") == {} + + +def test_backfill_batches_rather_than_truncating(monkeypatch): + """Batched at the indexer's own size, and every id is rebuilt. Measured + against the live PDB a rebuild is one round trip -- 500 ids in 1.5s + against 0.37s for a single id -- so batching is what keeps it cheap, and + a set larger than one batch is looped, never trimmed.""" + seen = {} + + class FakeIndexer: + def get_vfb_json_query(self, ids): + seen["ids"] = list(ids) + seen.setdefault("batches", []).append(len(ids)) + return "MATCH (n) RETURN n" + + def generate_solr_doc(self, result, request=None): + return {"id": result["term"]["core"]["short_form"], + "anat_query": {"set": '{"ok": true}'}} + + monkeypatch.setattr(tif, "_QUERY_FIELD_INDEXERS", {"anat_query": FakeIndexer}) + monkeypatch.setattr(tif, "_INDEXERS", {"class": object}) + monkeypatch.setattr(tif, "_write_query_docs", lambda docs, field: True) + + ids = ["FBbt_%07d" % i for i in range(tif.BACKFILL_BATCH_SIZE + 25)] + rows = [{"term": {"core": {"short_form": i}}} for i in ids] + import vfbquery.vfb_queries as vq + calls = {"n": 0} + + def cursor(r): + i = calls["n"]; calls["n"] += 1 + return rows[i * tif.BACKFILL_BATCH_SIZE:(i + 1) * tif.BACKFILL_BATCH_SIZE] \ + if i == 0 else rows[tif.BACKFILL_BATCH_SIZE:] + + monkeypatch.setattr(vq, "get_dict_cursor", lambda: cursor) + monkeypatch.setattr(vq, "vc", _StubConnect()) + + rebuilt = tif.backfill_query_field(ids, "anat_query") + # Batched, not truncated: the last batch is the 25-id remainder, and every + # id came back. Dropping the tail would return a quietly short answer, + # which is the thing this whole module exists to stop. + assert len(seen["ids"]) == 25 + assert seen["batches"] == [tif.BACKFILL_BATCH_SIZE, 25] + assert rebuilt[ids[0]] == '{"ok": true}' # payload returned, not re-read + + +def test_backfill_returns_payloads_for_the_current_request(monkeypatch): + """The indexer writes with commitWithin 60s, so re-reading SOLR in this + request would still miss. The payloads have to come back in memory.""" + class FakeIndexer: + def get_vfb_json_query(self, ids): + return "MATCH (n) RETURN n" + + def generate_solr_doc(self, result, request=None): + return {"id": "VFB_00107fob", + "anat_image_query": {"set": '{"term": {}}'}} + + monkeypatch.setattr(tif, "_QUERY_FIELD_INDEXERS", + {"anat_image_query": FakeIndexer}) + monkeypatch.setattr(tif, "_INDEXERS", {"class": object}) + written = [] + monkeypatch.setattr(tif, "_write_query_docs", + lambda docs, field: written.append((len(docs), field)) or True) + import vfbquery.vfb_queries as vq + monkeypatch.setattr(vq, "get_dict_cursor", lambda: (lambda r: [{"x": 1}])) + monkeypatch.setattr(vq, "vc", _StubConnect()) + + out = tif.backfill_query_field(["VFB_00107fob"], "anat_image_query") + assert out == {"VFB_00107fob": '{"term": {}}'} + assert written == [(1, "anat_image_query")] + + +def test_a_failed_rebuild_returns_nothing_rather_than_raising(monkeypatch): + class FakeIndexer: + def get_vfb_json_query(self, ids): + return "MATCH (n) RETURN n" + + monkeypatch.setattr(tif, "_QUERY_FIELD_INDEXERS", {"anat_query": FakeIndexer}) + monkeypatch.setattr(tif, "_INDEXERS", {"class": object}) + import vfbquery.vfb_queries as vq + + def boom(*a, **k): + raise RuntimeError("neo4j down") + + monkeypatch.setattr(vq, "get_dict_cursor", lambda: boom) + monkeypatch.setattr(vq, "vc", _StubConnect()) + assert tif.backfill_query_field(["FBbt_00003748"], "anat_query") == {} + + +def test_query_writes_honour_the_cache_guard(monkeypatch): + monkeypatch.setenv("VFBQUERY_CACHE_ENABLED", "false") + monkeypatch.setattr(tif, "_INDEXERS", {"class": object}) + sent = [] + monkeypatch.setattr(tif, "_send_solr_docs", + lambda d, f: sent.append(f) or True) + assert tif._write_query_docs([{"id": "x"}], "anat_query") is False + assert sent == [] diff --git a/src/test/test_vfb_connectivity.py b/src/test/test_vfb_connectivity.py index b285cf3..2cf7191 100644 --- a/src/test/test_vfb_connectivity.py +++ b/src/test/test_vfb_connectivity.py @@ -301,7 +301,12 @@ def test_default_excludes_are_a_strict_subset_of_everything(self, expansion_resu group_by_class=True, ) assert DEFAULT_EXCLUDE_DBS == ["hb", "fafb"] - assert 0 < expansion_result["count"] < everything["count"] + # Subset, not strict subset. Excluding a dataset can only ever remove + # rows, and that is the contract worth asserting. Demanding a strict + # decrease additionally requires hemibrain or FAFB to hold rows for + # *this* term pair, which is backend content -- it was true when this + # was written and is not true now (17 either way). + assert 0 < expansion_result["count"] <= everything["count"] @pytest.mark.integration def test_default_matches_passing_it_explicitly(self): diff --git a/src/vfbquery/term_info_fallback.py b/src/vfbquery/term_info_fallback.py index a001609..e5a2bfb 100644 --- a/src/vfbquery/term_info_fallback.py +++ b/src/vfbquery/term_info_fallback.py @@ -1,4 +1,4 @@ -"""Build a missing ``term_info`` SOLR document on demand. +"""Build missing ``vfb_json`` SOLR content on demand. ``get_term_info`` reads a pre-built ``term_info`` document out of the ``vfb_json`` SOLR collection. Those documents are written in bulk by @@ -55,6 +55,7 @@ class page as well. _IMPORT_ERROR = None _INDEXERS = None +_QUERY_FIELD_INDEXERS = None _send_solr_docs = None _import_lock = threading.Lock() @@ -101,21 +102,45 @@ def schema_version(): return "unpinned" +def _split_solr_url(url): + """Split a pysolr collection URL into the indexer's two env vars. + + ``http://host/solr/vfb_json/`` -> ``("http://host/solr", "vfb_json")``. + """ + if not url: + return None, None + base = url.rstrip("/") + server, _, collection = base.rpartition("/") + if not server or not collection: + return None, None + return server, collection + + def _seed_indexer_env(): - """Point the indexer's Neo4j connection at the one VFBquery already uses. + """Point the indexer at the same PDB and SOLR that VFBquery already uses. ``BaseQueryIndexer.__init__`` reads PDBserver/PDBuser/PDBpassword from the - environment and builds its own connection. Rather than run the fallback - against a different database than the rest of the process, fill those in - from VFBquery's own client when the deployment has not set them. Anything - already in the environment wins. + environment and builds its own Neo4j connection, and ``solr_client`` + builds its update URL from SOLRserver/SOLRcollection. None of those are + set in a VFBquery deployment, so fill them from VFBquery's own clients + rather than running the fallback against a different database -- or, in + the SOLR case, against nothing at all: ``get_solr_update_url`` returns + None when they are unset and ``send_solr_payload`` then returns False + without a request, so every rebuilt document was served but never + indexed and the next caller paid the same Neo4j round trip again. + + Anything already in the environment wins, so a deployment can still + point the fallback somewhere else. """ - from .vfb_queries import vc + from .vfb_queries import vc, vfb_solr nc = vc.nc + solr_server, solr_collection = _split_solr_url(getattr(vfb_solr, "url", None)) defaults = { "PDBserver": getattr(nc, "base_uri", None), "PDBuser": getattr(nc, "usr", None), "PDBpassword": getattr(nc, "pwd", None), + "SOLRserver": solr_server, + "SOLRcollection": solr_collection, } for key, value in defaults.items(): if value and not os.getenv(key): @@ -164,7 +189,7 @@ def _indexer_importable(): def _load_indexers(): """Import the indexer classes once, or record why we could not.""" - global _INDEXERS, _send_solr_docs, _IMPORT_ERROR + global _INDEXERS, _QUERY_FIELD_INDEXERS, _send_solr_docs, _IMPORT_ERROR if _INDEXERS is not None or _IMPORT_ERROR is not None: return with _import_lock: @@ -192,6 +217,15 @@ def _load_indexers(): SplitClassTermInfoQueryIndexer) from src.indexers.term_info.template_term_info_indexer import ( TemplateTermInfoQueryIndexer) + from src.indexers.anat_query_indexer import AnatQueryIndexer + from src.indexers.anat_image_query_indexer import ( + AnatImageQueryIndexer) + from src.indexers.connectivity.neuron_upstream_connectivity_indexer import ( + NeuronUpstreamConnectivityIndexer) + from src.indexers.connectivity.neuron_downstream_connectivity_indexer import ( + NeuronDownstreamConnectivityIndexer) + from src.indexers.scRNAseq.cluster_expression_query_indexer import ( + ClusterExpressionQueryIndexer) from src.solr_client import send_solr_docs except Exception as e: # ImportError, or a missing env var _IMPORT_ERROR = e @@ -207,6 +241,13 @@ def _load_indexers(): "split_class": SplitClassTermInfoQueryIndexer, "class": ClassTermInfoQueryIndexer, } + _QUERY_FIELD_INDEXERS = { + "anat_query": AnatQueryIndexer, + "anat_image_query": AnatImageQueryIndexer, + "upstream_connectivity_query": NeuronUpstreamConnectivityIndexer, + "downstream_connectivity_query": NeuronDownstreamConnectivityIndexer, + "cluster_expression": ClusterExpressionQueryIndexer, + } _send_solr_docs = send_solr_docs @@ -383,3 +424,101 @@ def backfill_term_info(short_form): else: print("term_info fallback: built %s but did not index it" % short_form) return payload + + +# -------------------------------------------------------------------------- +# Per-field query indexes +# -------------------------------------------------------------------------- + +#: Ids per rebuild request -- the indexer's own REQUEST_BATCH_SIZE. This is a +#: batch size, not a ceiling: a missing set larger than this is rebuilt in +#: successive batches rather than truncated, because an individual call +#: fixing itself slowly beats one that returns a quietly incomplete answer. +#: Measured against the live PDB a rebuild costs one round trip rather than +#: per-id work -- 1 id 0.37 s, 10 ids 1.41 s, 50 ids 0.91 s, 200 ids 0.80 s, +#: 500 ids 1.49 s (3 ms/id) -- so the batching is what keeps that cheap. +BACKFILL_BATCH_SIZE = 500 + + +def query_field_fallback_available(solr_field): + """True when a missing ``solr_field`` index can be rebuilt.""" + _load_indexers() + return bool(_QUERY_FIELD_INDEXERS) and solr_field in _QUERY_FIELD_INDEXERS + + +def backfill_query_field(short_forms, solr_field): + """Rebuild a precomputed query index for ids SOLR has no entry for. + + These fields are written by the same bulk indexer as ``term_info`` and go + missing the same way, but they fail far more quietly: both hydration paths + skip an id whose document lacks the field, so the rows and the count come + back short with nothing to say so. Measured on the live index the shortfall + is 0.5% of all images and about 2% of connectivity classes -- but it is + concentrated where people look, with two thirds of the instances of + medulla, antennal lobe and mushroom body missing ``anat_image_query``. + + Returns the rebuilt payloads so the *current* request can use them: the + indexer writes with ``commitWithin`` 60s, so re-reading SOLR here would + still miss. + + :param short_forms: ids with no usable entry for this field + :param solr_field: the SOLR field, which is also the indexer's service name + :return: ``{short_form: field_payload_json}`` for whatever was rebuilt + """ + if not short_forms: + return {} + if not query_field_fallback_available(solr_field): + _warn_unavailable_once() + return {} + + from .vfb_queries import vc, get_dict_cursor + + ids = list(short_forms) + indexer = _QUERY_FIELD_INDEXERS[solr_field]() + if len(ids) > BACKFILL_BATCH_SIZE: + print("query fallback: %d ids missing %s, rebuilding in %d batches" + % (len(ids), solr_field, + -(-len(ids) // BACKFILL_BATCH_SIZE))) + + rebuilt = {} + for start in range(0, len(ids), BACKFILL_BATCH_SIZE): + batch = ids[start:start + BACKFILL_BATCH_SIZE] + try: + rows = get_dict_cursor()(vc.nc.commit_list([ + indexer.get_vfb_json_query(batch)])) + except Exception as e: + print("query fallback: rebuilding %s for %d ids failed: %s" + % (solr_field, len(batch), e)) + continue + + solr_docs = [] + for result in rows or []: + try: + doc = indexer.generate_solr_doc(result, request=None) + except Exception as e: + print("query fallback: could not shape a %s document: %s" + % (solr_field, e)) + continue + solr_docs.append(doc) + rebuilt[doc["id"]] = doc[solr_field]["set"] + + if solr_docs: + _write_query_docs(solr_docs, solr_field) + + print("query fallback: rebuilt %d/%d missing %s entries" + % (len(rebuilt), len(ids), solr_field)) + return rebuilt + + +def _write_query_docs(solr_docs, solr_field): + """Index rebuilt query documents, honouring the same guard as term_info.""" + from .solr_result_cache import solr_caching_disabled + if solr_caching_disabled(): + print("query fallback: cache disabled, not writing %d %s entries" + % (len(solr_docs), solr_field)) + return False + try: + return bool(_send_solr_docs(solr_docs, solr_field)) + except Exception as e: + print("query fallback: SOLR write failed for %s: %s" % (solr_field, e)) + return False diff --git a/src/vfbquery/vfb_queries.py b/src/vfbquery/vfb_queries.py index 9fd2aa8..eb86436 100644 --- a/src/vfbquery/vfb_queries.py +++ b/src/vfbquery/vfb_queries.py @@ -1,6 +1,6 @@ import pysolr from .term_info_queries import deserialize_term_info -from .term_info_fallback import backfill_term_info +from .term_info_fallback import backfill_query_field, backfill_term_info # Replace VfbConnect import with our new SimpleVFBConnect from .owlery_client import SimpleVFBConnect # Keep dict_cursor if it's used elsewhere - lazy import to avoid GUI issues @@ -4427,20 +4427,31 @@ def _fetch_connectivity_entries(short_form: str, solr_field: str, subclass_ids=N results = vfb_solr.search( q='id:*', fq=f'{{!terms f=id}}{id_list}', - fl=solr_field, + fl=f'id,{solr_field}', rows=len(subclass_ids), ) except Exception as e: print(f"Error querying Solr for {solr_field}: {e}") return [] + # Same silent gap as the Owlery hydration path: a class with no entry for + # this field contributed nothing and said nothing. Roughly 2% of the + # connectivity population is in that state at any time. Rebuild in one + # batched call rather than returning a short answer. + field_by_id = {} + for doc in results.docs: + if solr_field in doc: + raw = doc[solr_field] + field_by_id[doc['id']] = raw[0] if isinstance(raw, list) else raw + missing_ids = [sid for sid in subclass_ids if sid not in field_by_id] + if missing_ids: + print(f"{solr_field}: {len(missing_ids)}/{len(subclass_ids)} ids " + f"have no entry; rebuilding") + field_by_id.update(backfill_query_field(missing_ids, solr_field)) + # Step 3: Parse all connectivity JSON from all returned docs all_entries = [] - for doc in results.docs: - if solr_field not in doc: - continue - raw = doc[solr_field] - field_json = raw[0] if isinstance(raw, list) else raw + for field_json in field_by_id.values(): try: entries = json.loads(field_json) except (json.JSONDecodeError, TypeError): @@ -4590,6 +4601,32 @@ def _bulk_fetch_per_instance_connectivity(instance_ids): except Exception as e: print(f"Failed to parse cached connectivity for {iid}: {e}") missing = [i for i in instance_ids if i not in found] + if missing: + # A miss here is not a missing bulk index -- it is VFBquery's own + # result cache never having been asked for this instance. Skipping it + # made connected_n / pairwise_connections / total_weight quietly low, + # which the caller could only describe as "a slight underestimate". + # Compute them instead: get_neuron_neuron_connectivity is wrapped in + # @with_solr_cache, so each call also fills the cache and the next + # caller pays nothing. This is deliberately unbounded -- an individual + # call should fix itself even if that call is slow, rather than return + # a number that is wrong without saying so. + print(f"Computing per-instance connectivity for {len(missing)} " + f"uncached instance(s); this call will be slower.") + still_missing = [] + for iid in missing: + try: + result = get_neuron_neuron_connectivity( + iid, return_dataframe=False) + except Exception as e: + print(f"Live per-instance connectivity failed for {iid}: {e}") + still_missing.append(iid) + continue + if isinstance(result, dict): + found[iid] = result.get('rows', []) + else: + still_missing.append(iid) + missing = still_missing return found, missing @@ -5391,17 +5428,31 @@ def _owlery_query_to_results(owl_query_string: str, short_form: str, return_data results = vfb_solr.search( q='id:*', fq=f'{{!terms f=id}}{id_list}', - fl=solr_field, + fl=f'id,{solr_field}', rows=len(class_ids) ) - - # Process all results + + # An id whose document is missing, or which has a document without + # this field, used to be skipped in silence -- the rows AND the + # count came back short with nothing to say so. Both happen for + # any record added since the bulk indexer last completed. Rebuild + # them from the indexer's own query in one batched call: the cost + # is a single round trip (500 ids in ~1.5s), so this is cheaper + # than the Owlery expansion the query has already paid for. + field_by_id = {} for doc in results.docs: - if solr_field not in doc: - continue - + if solr_field in doc: + raw = doc[solr_field] + field_by_id[doc['id']] = raw[0] if isinstance(raw, list) else raw + missing_ids = [cid for cid in class_ids if cid not in field_by_id] + if missing_ids: + print(f"{solr_field}: {len(missing_ids)}/{len(class_ids)} ids " + f"have no entry; rebuilding") + field_by_id.update(backfill_query_field(missing_ids, solr_field)) + + # Process all results + for field_data_str in field_by_id.values(): # Parse the SOLR field JSON string - field_data_str = doc[solr_field][0] field_data = json.loads(field_data_str) # Extract core term information @@ -7551,12 +7602,27 @@ def _build_descendants_subclass(root_id): results = vfb_solr.search( q='id:*', fq=f'{{!terms f=id}}{id_list}', fl='id,term_info', rows=len(all_desc) ) + # Collect first, so a term with no document can be rebuilt rather + # than dropped. Skipping it here did not just lose its own node -- + # it lost the whole branch beneath it, because nothing was ever + # appended to its parent's child list. + term_info_by_id = {} for doc in results.docs: - child_id = doc.get('id', '') - if 'term_info' not in doc: - continue - raw = doc['term_info'] - ti = json.loads(raw[0] if isinstance(raw, list) else raw) + if 'term_info' in doc: + raw = doc['term_info'] + term_info_by_id[doc.get('id', '')] = ( + raw[0] if isinstance(raw, list) else raw) + missing_ids = [d for d in all_desc if d not in term_info_by_id] + if missing_ids: + print(f"hierarchy: {len(missing_ids)}/{len(all_desc)} terms have " + f"no term_info document; rebuilding") + for mid in missing_ids: + payload = backfill_term_info(mid) + if payload: + term_info_by_id[mid] = payload + + for child_id, raw_term_info in term_info_by_id.items(): + ti = json.loads(raw_term_info) parents_in_tree = [p['short_form'] for p in ti.get('parents', []) if p['short_form'] in tree_ids] if parents_in_tree: for pid in parents_in_tree: