From b77f6a8d3090c1dbe6b0150a9089e29f564997bc Mon Sep 17 00:00:00 2001 From: Robbie Court Date: Fri, 4 Sep 2026 10:44:24 +0000 Subject: [PATCH 1/3] Rebuild missing query indexes too, and give the fallback a SOLR endpoint Three things, all downstream of the same cause: records that reach the PDB after the bulk indexer last completed have nothing in vfb_json, and the `precompute live query results` job has not completed since 22 June. ## The query indexes went missing as silently as term_info did Both hydration paths skipped an id whose document lacked the field, so the rows AND the count came back short with nothing to say so. For the connectivity queries that is not "fewer rows" -- it is a confident empty answer. Measured before and after on the live backend: FBbt_00110929 alpha/beta core Kenyon cell 0 rows -> 236 FBbt_00048273 wing bristle mechanosensory 0 rows -> 604 152 of 4,000 sampled connectivity classes are in that state. A user asking what is downstream of a Kenyon cell was being told, with a straight face, nothing. Both call sites now collect what SOLR returned, work out which requested ids have no usable entry, and rebuild those in one batched call through the indexer's own query -- the same borrowed-not-copied approach as term_info. Measured on the live PDB a rebuild costs one round trip rather than per-id work: 1 id 0.37s, 50 ids 0.91s, 200 ids 0.80s, 500 ids 1.49s (3 ms/id). So the whole missing set goes at once, capped at the indexer's own REQUEST_BATCH_SIZE of 500 purely to stop a pathological expansion stalling. The rebuilt payloads are returned in memory and used by the current request, because the indexer writes with commitWithin 60s and re-reading SOLR here would still miss. Global coverage measured against each indexer's own parameter query: anat_image_query 0.5% missing, up/downstream connectivity ~2.1%, cluster_expression 0%, anat_query has 279 MORE documents than the population (stale entries a fallback cannot fix -- only the bulk job clearing them can). ## The term_info fallback never actually indexed anything solr_client builds its update URL from SOLRserver/SOLRcollection, and a VFBquery deployment sets neither, so get_solr_update_url() returned None and send_solr_payload returned False without making a request. Confirmed against live 1.22.49: get_term_info("VFB_00107fob") served a complete document while id:VFB_00107fob stayed numFound 0 past the 60s commitWithin. Both are now seeded from VFBquery's own pysolr client alongside the PDB variables. The tests missed it because the one write test set VFBQUERY_CACHE_ENABLED false to prove the guard, and that guard short-circuits before the URL is needed -- the path that would have caught it was the path the test disabled. ## Two tests that asserted content, not contract term_info_serialization_dataset asserted exactly 4 types; DataSets gained stage labels and Ito2013 now has 5 (Adult). It asserts the types it is about instead. test_default_excludes_are_a_strict_subset_of_everything demanded a strict decrease from excluding hemibrain and FAFB, which additionally requires those datasets to hold rows for that particular term pair -- content, not contract. Now asserts subset, which is what excluding a dataset actually guarantees. Note on scope: the anat_query / anat_image_query rebuilds are wired but latent. The expansions those queries produce did not contain unindexed ids in any case I could construct -- get_instances and get_images_neurons are unchanged before and after. The demonstrated win is the connectivity path. --- src/test/term_info_queries_test.py | 8 +- src/test/test_term_info_fallback.py | 157 ++++++++++++++++++++++++++++ src/test/test_vfb_connectivity.py | 7 +- src/vfbquery/term_info_fallback.py | 152 +++++++++++++++++++++++++-- src/vfbquery/vfb_queries.py | 53 +++++++--- 5 files changed, 352 insertions(+), 25 deletions(-) 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..6475023 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,115 @@ 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_and_caps(monkeypatch): + """One batched call, capped at the indexer's own batch size. Measured + against the live PDB a rebuild is one round trip -- 500 ids in 1.5s + against 0.37s for a single id -- so the whole missing set goes at once.""" + seen = {} + + class FakeIndexer: + def get_vfb_json_query(self, ids): + seen["ids"] = list(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.MAX_BACKFILL_IDS + 25)] + rows = [{"term": {"core": {"short_form": i}}} for i in ids[:tif.MAX_BACKFILL_IDS]] + import vfbquery.vfb_queries as vq + monkeypatch.setattr(vq, "get_dict_cursor", lambda: (lambda r: rows)) + monkeypatch.setattr(vq, "vc", _StubConnect()) + + rebuilt = tif.backfill_query_field(ids, "anat_query") + assert len(seen["ids"]) == tif.MAX_BACKFILL_IDS # capped + assert len(rebuilt) == tif.MAX_BACKFILL_IDS + 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..6704ae7 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,98 @@ 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 +# -------------------------------------------------------------------------- + +#: How many ids one rebuild will attempt. This is the indexer's own +#: REQUEST_BATCH_SIZE, and it is a cap rather than a target: 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 whole missing set goes in a single batched call and the +#: cap exists only to stop a pathological expansion turning into a stall. +#: For scale: the worst real gap found in one query expansion was 8 ids. +MAX_BACKFILL_IDS = 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) + if len(ids) > MAX_BACKFILL_IDS: + print("query fallback: %d ids missing %s, rebuilding the first %d" + % (len(ids), solr_field, MAX_BACKFILL_IDS)) + ids = ids[:MAX_BACKFILL_IDS] + + indexer = _QUERY_FIELD_INDEXERS[solr_field]() + try: + rows = get_dict_cursor()(vc.nc.commit_list([ + indexer.get_vfb_json_query(ids)])) + except Exception as e: + print("query fallback: rebuilding %s for %d ids failed: %s" + % (solr_field, len(ids), e)) + return {} + + rebuilt = {} + 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..b89c3fb 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): @@ -5391,17 +5402,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 From 3ea8fa7f04e553d977fd58aac1fed476d072b7f8 Mon Sep 17 00:00:00 2001 From: Robbie Court Date: Fri, 4 Sep 2026 11:01:50 +0000 Subject: [PATCH 2/3] Fix uncached per-instance connectivity instead of skipping it _bulk_fetch_per_instance_connectivity skipped instances with no cached result and warned that connected_n / pairwise_connections / total_weight would be "a slight underestimate". That is a wrong number presented as a real one. This miss is a different animal to the others here: it is not a gap in the bulk index but VFBquery's own @with_solr_cache never having been asked for that instance. So the fix is simply to ask -- get_neuron_neuron_connectivity is wrapped in the cache decorator, so computing a missing instance also fills the cache and the next caller pays nothing. Deliberately unbounded. An individual call should fix itself even if that call is slow, rather than return a quietly incomplete count; misses are rare enough for that to be safe (0 of 4,000 connectivity individuals sampled had no cached result, and the case that surfaced this was 1 of 794 instances under FBbt_00048273). Same principle applied to the query-index rebuild: BACKFILL_BATCH_SIZE replaces MAX_BACKFILL_IDS, and a missing set larger than one batch is now looped rather than truncated. Batching is what makes it cheap -- 500 ids in 1.49s against 0.37s for one -- so there is no reason to drop the tail. Verified live: FBbt_00048273 logs "Computing per-instance connectivity for 1 uncached instance(s)" and still returns its 604 rows. --- src/test/test_term_info_fallback.py | 28 +++++++++---- src/vfbquery/term_info_fallback.py | 61 +++++++++++++++-------------- src/vfbquery/vfb_queries.py | 26 ++++++++++++ 3 files changed, 78 insertions(+), 37 deletions(-) diff --git a/src/test/test_term_info_fallback.py b/src/test/test_term_info_fallback.py index 6475023..4e4d1f1 100644 --- a/src/test/test_term_info_fallback.py +++ b/src/test/test_term_info_fallback.py @@ -239,15 +239,17 @@ def explode(*a, **k): assert tif.backfill_query_field([], "anat_query") == {} -def test_backfill_batches_and_caps(monkeypatch): - """One batched call, capped at the indexer's own batch size. Measured +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 the whole missing set goes at once.""" + 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): @@ -258,15 +260,25 @@ def generate_solr_doc(self, result, request=None): 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.MAX_BACKFILL_IDS + 25)] - rows = [{"term": {"core": {"short_form": i}}} for i in ids[:tif.MAX_BACKFILL_IDS]] + 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 - monkeypatch.setattr(vq, "get_dict_cursor", lambda: (lambda r: rows)) + 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") - assert len(seen["ids"]) == tif.MAX_BACKFILL_IDS # capped - assert len(rebuilt) == tif.MAX_BACKFILL_IDS + # 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 diff --git a/src/vfbquery/term_info_fallback.py b/src/vfbquery/term_info_fallback.py index 6704ae7..e5a2bfb 100644 --- a/src/vfbquery/term_info_fallback.py +++ b/src/vfbquery/term_info_fallback.py @@ -430,14 +430,14 @@ def backfill_term_info(short_form): # Per-field query indexes # -------------------------------------------------------------------------- -#: How many ids one rebuild will attempt. This is the indexer's own -#: REQUEST_BATCH_SIZE, and it is a cap rather than a target: 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 whole missing set goes in a single batched call and the -#: cap exists only to stop a pathological expansion turning into a stall. -#: For scale: the worst real gap found in one query expansion was 8 ids. -MAX_BACKFILL_IDS = 500 +#: 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): @@ -474,34 +474,37 @@ def backfill_query_field(short_forms, solr_field): from .vfb_queries import vc, get_dict_cursor ids = list(short_forms) - if len(ids) > MAX_BACKFILL_IDS: - print("query fallback: %d ids missing %s, rebuilding the first %d" - % (len(ids), solr_field, MAX_BACKFILL_IDS)) - ids = ids[:MAX_BACKFILL_IDS] - indexer = _QUERY_FIELD_INDEXERS[solr_field]() - try: - rows = get_dict_cursor()(vc.nc.commit_list([ - indexer.get_vfb_json_query(ids)])) - except Exception as e: - print("query fallback: rebuilding %s for %d ids failed: %s" - % (solr_field, len(ids), e)) - return {} + 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 = {} - solr_docs = [] - for result in rows or []: + for start in range(0, len(ids), BACKFILL_BATCH_SIZE): + batch = ids[start:start + BACKFILL_BATCH_SIZE] try: - doc = indexer.generate_solr_doc(result, request=None) + rows = get_dict_cursor()(vc.nc.commit_list([ + indexer.get_vfb_json_query(batch)])) except Exception as e: - print("query fallback: could not shape a %s document: %s" - % (solr_field, e)) + print("query fallback: rebuilding %s for %d ids failed: %s" + % (solr_field, len(batch), e)) continue - solr_docs.append(doc) - rebuilt[doc["id"]] = doc[solr_field]["set"] - if solr_docs: - _write_query_docs(solr_docs, solr_field) + 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 diff --git a/src/vfbquery/vfb_queries.py b/src/vfbquery/vfb_queries.py index b89c3fb..cfd878a 100644 --- a/src/vfbquery/vfb_queries.py +++ b/src/vfbquery/vfb_queries.py @@ -4601,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 From 6ca82bfa5556a607640aa5365b9b064ab99ade64 Mon Sep 17 00:00:00 2001 From: Robbie Court Date: Fri, 4 Sep 2026 14:01:36 +0000 Subject: [PATCH 3/3] Rebuild missing term_info in the hierarchy instead of dropping the branch Fourth hydration path with the same silent skip. get_hierarchy batch-reads term_info from SOLR to work out each term's parents and did `continue` on a document that was not there. That did not just lose the term's own node: it lost everything beneath it, because nothing was ever appended to its parent's child list, so an unindexed term took its whole branch out of the tree with no error and no warning. Collect the documents first, rebuild any that are missing through the term_info fallback, then build the tree from the merged set. --- src/vfbquery/vfb_queries.py | 25 ++++++++++++++++++++----- 1 file changed, 20 insertions(+), 5 deletions(-) diff --git a/src/vfbquery/vfb_queries.py b/src/vfbquery/vfb_queries.py index cfd878a..eb86436 100644 --- a/src/vfbquery/vfb_queries.py +++ b/src/vfbquery/vfb_queries.py @@ -7602,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: