Patch delays - #142
Open
YaphetKG wants to merge 46 commits into
Open
Conversation
YaphetKG
requested review from
vladimir2217 and
yskale
and removed request for
vladimir2217
June 16, 2026 20:19
vladimir2217
approved these changes
Jun 17, 2026
Dug's data classes moved into the dug_data_model library, and the module they used to live in (dug.core.parsers._base) no longer imports. jsonpickle does not raise on that -- it returns the raw dict -- so artifacts written by an older dug index as dicts and fail with "'dict' object has no attribute 'id'". With incremental ingestion those artifacts are only rewritten when their source changes, so they can sit in lakefs across dug upgrades indefinitely. scripts/migrate_pickled_classes.py restamps them in place instead of re-annotating: --scan reports which stored classes no longer import and whether any stored field is missing from the current model, --fix decodes through a legacy module alias, fills in fields added since, and re-encodes. Also import random and time in pipelines/base.py -- init_annotator's retry path used both, so a transient annotator failure raised NameError instead of retrying.
README described a codebase that no longer exists: a KGX fork install, a `bin/roger all` entrypoint, `dags/roger/config/`, Python 3.7, and ~450 lines of pasted CLI output. Replaced with an orientation doc -- what Roger produces, the two workflows and how they differ, an annotated repo tour with a reading order, quickstart, how to run tests without host dependencies, config, the concepts that bite early (lakefs as source of truth, incremental runs, ES as a derived index, jsonpickle-encoded artifacts), and a troubleshooting table. Kept the KGX merge/schema type-conflict rules, the bulk-loader CSV grouping, and the k8s/Helm prerequisites. Dropped the stale run log and screenshots. CLAUDE.md covers what contributors need but the README should not carry: the incremental state machine and why the success callback must not re-resolve refs, deletion propagation, lakefs task wiring flags, the DAG shapes, and the dug_data_model migration runbook.
trying out seperating indexing jobs and adding removal steps for new …
… live in smaller foot print
… over files
Annotation dominates the cost of annotate_and_index, and three separate
things were making it far more expensive than the work it actually does.
Cache POST responses. dug builds the annotation session with
requests_cache, whose allowable_methods defaults to ('GET', 'HEAD').
Only node normalization is a GET; token classification, sapbert, and
name-resolution synonyms are POSTs and so were never cached. Since the
request body is part of requests_cache's key for POST, and every one of
these services is a read-only lookup keyed entirely by that body,
caching them is correct rather than a heuristic. This matters most for
dbGaP-shaped data, where the parsers emit the study element into every
one of a study's data-dict files: bdc-parent re-annotated the same 24
study descriptions across 61,597 files, at ~53s each against ~1.4s for
the variable the file actually contributes.
http_cache_expire_seconds also sets expire_after, which dug never did.
requests_cache's redis backend only writes entries with SETEX when an
expiry is set, so this keeps annotation cache keys volatile while the
FalkorDB graph keys sharing that redis stay permanent -- redis can then
be given a maxmemory with volatile-lru and will evict cache before it
touches the graph. It also bounds the normalizer GETs, which dug was
caching with no expiry at all.
Resume mid-annotation. Task output only reaches lakefs on task success,
so a task that died at file 40,000 of 61,597 discarded all of it and the
retry restarted at zero. annotation_is_complete now skips input files
whose elements.txt and concepts.txt both exist and are non-empty (both,
because they are written in sequence and a kill between them leaves an
unusable directory); clean_up keeps the output dir on failure; and
reuse_prior_try_outputs hard-links earlier tries' output into the current
try, which is needed because generate_dir_name_from_task_instance stamps
the try number into the path. Successful commits then clean up all tries.
This covers retries within a dag run, not a fresh dag run.
Thread annotate_files over input files (annotate_workers, default 4).
Files are wholly independent -- own parse, own Crawler, own output dir --
and the work is nearly all HTTP wait, so this scales despite the GIL.
Each worker gets its own session and annotator; the response cache is
shared, so workers still see each other's annotations. Element-level
concurrency would have to live in dug's Crawler.
Also fixes clean_up, which had its input and output suffixes swapped.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
The `test` graph in the kebedey namespace has 3,889,306 nodes and zero relationships despite BulkLoad, Validate and CheckTranql all reporting success. Two independent defects combined to produce that. BulkLoad read the branch before its input existed. Task output only reaches lakefs in avalon_commit_callback, an on_success_callback, while Airflow releases downstream tasks on task state -- so an upstream edge guarantees the producer finished, not that its output is visible. That window is normally negligible, but CreateBulkLoadEdges wrote ~12GB (one subclass_of csv is 11.7GB), so the upload and merge outlasted it: 16:52:57Z BulkLoad resolves main tip -> c400de2f (no edge objects) 16:54:37Z CreateBulkLoadEdges' commit lands on main 17:01:54Z BulkLoad exits 0 The narrow fix here is incremental_pull=False. BulkLoad deletes the graph and reloads it whole, so it always needs the complete node and edge csv set; an incremental pull hands it only what changed and it rebuilds the graph from that fragment. This is the same reasoning already applied to the elasticsearch task groups in annotate_and_index, which BulkLoad has the same wholesale-rebuild semantics as. It also matters immediately: the recorded state for this task is c400de2f, so re-running it as-is would diff to a set containing the edges but not the unchanged node csvs, wipe the graph, and try to load edges with no nodes. Nothing turned red because insert() only passes -R arguments when edge csvs are present, and a nodes-only load is a perfectly valid loader invocation. It now refuses to bulk load an edgeless graph when node csvs are present, so a build that loses its edges fails where it breaks. This does not close the underlying race, which affects every task that commits in a success callback. Committing output before the task reports success is the real fix and is deliberately left out of scope here. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Airflow records end_date and releases downstream tasks before it runs success callbacks, so wiring avalon_commit_callback as an on_success_callback meant every task advertised completion while its output was still uploading to lakefs. Downstream then pulled a branch that did not yet contain it. Both DAGs were losing data to this. In knowledge_graph_build, CreateBulkLoadEdges ended at 16:52:52 and BulkLoad started 7s later, but the 12GB of edge csvs did not land until 16:54:37 -- 105s after the task was declared done -- so BulkLoad loaded 3.9M nodes with zero relationships and exited 0. In annotate_and_index, annotate_bdc-topmed_files ended at 21:17:37 and its commit landed at 21:25:52, while make_kgx started at 21:17:49 and built kgx from commit 8ce02a35, 4.8 hours stale. Every one of those tasks was green. post_execute is the right hook: airflow calls it inside _execute_task immediately after execute() and before the SucceedTask message, so the commit lands before downstream is eligible, and an exception there fails the task rather than being swallowed. Swallowing is the second half of this. _run_task_state_change_callbacks only logs exceptions raised by callbacks, and avalon_commit_callback logged and dropped merge failures of its own -- after which the clean_up at the end of the callback deleted the local output. A failed merge therefore destroyed the work and still reported success. The merge failure is now re-raised, which fails the task and leaves the output in place for the retry to resume from, since on_failure_callback already passes keep_output=True. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
BulkLoad ran at the chart default 2Gi. That was survivable while it was silently loading nodes only -- 0.9GiB of csvs -- but the real load is 13.6GiB of edge csvs, about 78M edges, and the falkordb bulk loader keeps a node-identifier to internal-id map in memory for the whole graph so it can resolve edge endpoints. At 3.9M nodes that map alone is a large fraction of 2Gi before any batching. 15Gi matches MergeNodes and CreateBulkLoadEdges, the other two tasks that handle the graph at full scale. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
`save 300 100000` triggers on write volume, and a bulk load is nothing but write volume, so bgsave forks repeatedly while 78M edges are going in. Copy-on-write on a graph this size can add most of it again on top of the resident set, and the headroom between redis maxmemory and the pod memory limit is not sized for that -- the pod gets OOMKilled and takes the half-loaded graph with it. A snapshot taken partway through a load is worthless anyway. The bulk csvs in lakefs are the source of truth and insert() starts by deleting the graph, so there is nothing worth persisting until the load finishes. Doing it in the task rather than by hand around the run is the point: the restore is in a finally, so a load that dies partway still puts snapshots back. Failing to set it is not fatal and only warns, since a redis that refuses CONFIG should not block a multi-hour load; failing to restore it logs at error, because that one leaves durability changed behind us. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Every full text index the bulk load tried to create failed:
Unable to create Full Text Search Index on Label: biolink.GeographicLocation
errMsg: Invalid input '.': expected ')'
errCtx: CREATE FULLTEXT INDEX FOR (e: biolink.GeographicLocation) ON (e.name)
falkordb interpolates the label straight into the pattern -- see
Graph._create_typed_index, `pattern = f"(e:{label})"` -- so a label
containing a dot has to arrive quoted, and every biolink label contains a
dot. The range indexes on the same line were already backticked and
succeeded; the two full text ones were not and never did.
The loader catches the ResponseError and only prints it, so the task
still reported success. The graph has been built without name or synonym
full text indexes for as long as this line has looked like this.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
adding batched annotator for nameres and node norm calls
The canary caught this: 6 of 183 bdc-heartfailure files came back with
CHEBI:3759 where the previous serial run had CHEBI:37941. Same identifier
count either way, so nothing was lost -- but one curie was no longer
being conflated.
The node normalizer does not default its flags the same way on both
verbs:
GET drug_chemical_conflate default=True
POST drug_chemical_conflate default=False
dug builds a GET url that sets conflate and description and says nothing
about the rest, so its lookups have always run with
drug_chemical_conflate=true. The batched POST omitted it and therefore
got false. conflate and include_taxa are the same shape of trap, both
defaulting true on GET, and the previous code sent false for anything
absent from the url.
So the flags are now built from the GET defaults and overridden by
whatever the url actually specifies, rather than defaulting to false.
Verified against the deployed normalizer: single GET and batched POST
agree on all probes including CHEBI:3759 -> CHEBI:37941.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
bdc-biolincc was OOMKilled at the chart default of 2Gi. That default was sized for the serial annotator. annotate_workers now runs four threads, each holding one whole parsed file, and biolincc's largest files carry 13,299 elements -- jsonpickle.encode also builds the entire output string in memory before anything is written, so the peak is four of those at once. annotation.annotate_memory (default 6Gi) so it can be tuned with ROGER_ANNOTATION_ANNOTATE__MEMORY without another push, and so it moves together with annotate_workers, which is what actually drives it. Note this multiplies against dag parallelism: twelve datasets at 6Gi exceeds the namespace quota, so tasks will queue rather than start. That is the intended behaviour (task_publish_max_retries is -1) but it does mean raising annotate_workers and annotate_memory together needs a look at the quota. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Five make_kgx tasks failed with
(400) {"message":"update branch main: no changes"}
after doing their work correctly -- curesc logged "Wrote 4705 and 28114
edges" immediately before failing. lakefs answers a merge that would
change nothing with a 400, and a task whose output is byte-identical to
what the branch already holds produces exactly that. Deterministic work
over unchanged input lands there every time.
This is a regression from commit 6aefe2f. Before it, the merge exception
was logged and dropped, so the task went green. Re-raising was right for
a genuine merge failure -- the clean_up that follows deletes the local
output, so swallowing one destroyed the work and still reported success
-- but it made this benign case fatal too.
So the no-changes case is now recognised and treated as what it is:
state still advances, because the input was consumed and the branch holds
the correct content. Everything else still raises.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
… output Two problems from the bdc run, both rooted in defaults that were sized for a smaller pipeline. make_kgx and crawl still ran at the chart default 2Gi. 19ecbac gave annotate its own limit but stopped there, and make_kgx_bdc-recover was then OOMKilled building kgx for a 6090-file dataset. crawl has the same shape -- it expands every concept through tranql and accumulates the answers -- and crawl_bdc-parent, the largest of the twelve, had not run yet. Both now take annotation.annotate_memory, so the three tasks that hold a dataset in memory move together. keep_output on failure is now opt-in, and only annotate opts in. It exists so a retry can resume, which works because annotation_is_complete skips input files whose output already exists. crawl and make_kgx have no such check and redo everything, so their retained output is never read again -- it is pure disk cost. That cost was not theoretical: three failed crawls held 47GB of unusable output on the shared 120GB volume, which is what filled it, and the volume filling is what had failed them. Keeping their output made the failure permanent instead of transient. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Incremental (delta) lakefs ingestion for the annotate/index DAG, plus fixes uncovered while testing it.
Changes
find_sibling_files()lists each changed file's lakefs directory and pulls anyGapExchange_*sibling not in the diff. dbGaP data dicts depend on that sibling for study name/description, so a delta carrying only the data dict would otherwise fail to parse.build_indexer_obj()/build_search_obj()no longer pass index-name lists; dug now reads index names from config (to_dug_conf()already supplies concepts/variables/studies/sections/kg).Tests
tests/unit/test_tasks_incremental.py— state keys, ref resolution, diff bucketing/prefix filter, skip-on-no-change, first-run full download, incremental diff download, GapExchange sibling pull, manual-override precedence, state callback.