Add SQLite accession index and bounded on-demand bucket retrieval

FEATURE_BACKLOG.md

@@ -2,7 +2,7 @@
 
 Track desired features that are not yet implemented. Add new ideas here and update each item's status as work progresses. These entries describe future work; implementation details and priorities remain open unless explicitly agreed.
 
-Last updated: 2026-09-29.
+Last updated: 2026-09-30.
 
 ## Overview
 
@@ -11,8 +11,8 @@ Last updated: 2026-09-29.
 | F001 | GenBank inference coverage map | Planned; not implemented |
 | F002 | Taxonomy map and navigation | Planned; not implemented |
 | F003 | Comparison with RefSeq annotations | Planned; not implemented |
-| F004 | Download annotations by accession, individually or in bulk | Planned; only individual sampled segment downloads exist |
-| F005 | Fast accession lookup and retrieval across the full dataset | Planned; only a local sample index exists |
+| F004 | Download annotations by accession, individually or in bulk | Partial; individual indexed segment downloads exist |
+| F005 | Fast accession lookup and retrieval across the full dataset | Prototype implemented for three remote objects; full coverage pending |
 
 ## F001 — GenBank inference coverage map
 
@@ -74,7 +74,7 @@ Decisions and dependencies for implementation:
 
 **Goal:** Make it easy for users to download annotations for the assemblies or contigs they are interested in.
 
-Current baseline: the prototype can download one selected sampled segment as Parquet. Accession-level and bulk downloads are future work.
+Current baseline: the prototype can download one selected indexed segment as Parquet, fetched from the bucket on demand. Filenames include assembly accession, record name, and coordinates. Complete accession-level and bulk downloads are future work.
 
 Desired behavior:
 
@@ -97,7 +97,9 @@ Decisions and dependencies for implementation:
 
 **Goal:** Keep accession searches and annotation retrieval responsive as the app expands beyond its bundled sample.
 
-Current baseline: the prototype indexes local sample metadata and reads selected Parquet row groups. It does not search the full bucket.
+Current baseline: [the remote prototype](README.md#remote-prototype-scope) uses a read-only SQLite metadata index over three published bucket objects, with 53 segments across three assemblies. It fetches selected row groups on demand, checks source content hashes, uses a bounded cache, and displays lookup/retrieval timings. The app supports both older whole-contig records and segmented records. It does not search the full bucket.
+
+Remaining work includes expanding index coverage, incremental refresh/publication, pagination beyond 200 matches, and reducing large-row-group transfer costs. `benchmark_remote.py` records first-load and cached performance for the current subset.
 
 Desired behavior:
 

README.md

@@ -8,10 +8,10 @@ sdk_version: 6.6.0
 python_version: '3.10'
 app_file: app.py
 pinned: false
-short_description: Explore sampled genome annotations by accession ID
+short_description: Find genome annotations with on-demand bucket reads
 ---
 
-Search a small, real subset of [HuggingFaceBio/genbank-annotations](https://huggingface.co/buckets/HuggingFaceBio/genbank-annotations) by assembly accession, contig accession, or the exact `source_key`. View strand-specific CDS probabilities, inspect provenance, and download original segment rows as Parquet.
+Search an indexed subset of [HuggingFaceBio/genbank-annotations](https://huggingface.co/buckets/HuggingFaceBio/genbank-annotations) by assembly accession, contig accession, or the exact `source_key`. A local SQLite index locates records; probability data stays in the bucket and loads on demand. View strand-specific CDS probabilities, inspect provenance, and download original segment rows as Parquet.
 
 Future features and open design decisions are tracked in the [feature backlog](FEATURE_BACKLOG.md).
 
@@ -24,20 +24,54 @@ pip install -r requirements.txt
 python app.py
 ```
 
-Open http://localhost:7860. The bundled sample works offline and requires no token at runtime.
+Open http://localhost:7860. Remote reads require a Hugging Face login (`hf auth login`) or an `HF_TOKEN` with bucket read access. For the original offline sample, run `GENBANK_DATA_MODE=sample python app.py`; that mode requires no token.
 
 ## Deploy to a Space
 
 The preview Space is [HuggingFaceBio/genbank-annotation-explorer](https://huggingface.co/spaces/HuggingFaceBio/genbank-annotation-explorer), with private visibility.
 
-To deploy or update it, upload this directory, including `data/sample.parquet` and `data/manifest.json`. The README metadata configures the runtime, following the [Gradio Spaces documentation](https://huggingface.co/docs/hub/spaces-sdks-gradio). Keep the Space private for the initial preview: the bundled annotations will be accessible to its viewers.
+To deploy or update it, upload this directory, including `data/catalog.sqlite`. Set `HF_TOKEN` as a Space secret with read access to the bucket. Include `data/sample.parquet` and `data/manifest.json` if offline sample mode is desired. Credentials stay on the server and must never be committed. The README metadata configures the runtime, following the [Gradio Spaces documentation](https://huggingface.co/docs/hub/spaces-sdks-gradio). Keep the Space private for the initial preview.
+
+## Remote prototype scope
+
+The current snapshot indexes **53 segments, 3 assemblies, and 195,493,177 bases** from three explicitly selected Parquet objects in fungi, protozoa, and vertebrate_other. The SQLite file is **86,016 bytes**. Building it fetched about **208 KB** of file ranges for metadata, rather than the approximately **402 MB** of source files. This subset does not represent the full bucket or complete assemblies.
+
+Examples beyond the original bundled sample:
+
+- `JBPJTW010000350.1` — fungi, a complete aligned contig.
+- `CM126017.1` — protozoa, an older whole-contig schema without explicit segment columns.
+- `CM057449.1` — Australian lungfish, the final indexed segment of a larger contig, with coordinates beginning at 1,800,000,000. Only this segment is in the index.
+
+The app reports lookup time, retrieval time, bytes fetched, and whether the selected data came from the bucket or cache. Byte counts measure returned file ranges, excluding HTTP overhead and retries. Retrieval time excludes plot aggregation and browser rendering.
+
+## Retrieval and storage
+
+- SQLite contains accession aliases, metadata, object paths and content hashes, row groups, and row positions. Probability arrays and sequence are excluded.
+- The Space opens the index read-only. Search uses indexed exact aliases, and results are capped at 200 with the total match count shown. Narrow to a contig accession when necessary; pagination is not yet implemented.
+- Remote reads use `HfFileSystem` and PyArrow, fetching the selected row group with all source columns so downloads preserve the original record.
+- An in-memory LRU cache holds at most 512 MiB of Arrow table buffers. Fills are serialized to avoid duplicate concurrent downloads. Temporary read/plot buffers require additional memory; this is not a total process memory limit.
+- Row groups with more than 512 MiB of uncompressed Parquet data are rejected before decoding. Larger source groups will need smaller chunks or another retrieval path.
+- On a cache miss, the object content hash is checked before and after reading. Changed objects are rejected with an index-refresh message. Cached data remains tied to the indexed snapshot. Objects are not versioned by this app; replacement of a source requires rebuilding the index.
+- Older whole-contig records derive display coordinates `[0, aligned_bp_length)`. Their downloaded Parquet rows retain the original schema.
+- Authentication and retrieval errors are shown separately from accession misses. A failed segment selection clears the previous plot and download.
+
+## Rebuild the remote index and benchmark
+
+```bash
+python build_index.py
+python benchmark_remote.py
+```
+
+The index builder reads only metadata columns from the three default files. Use repeated `--source annotations/...parquet` arguments for another bounded set of up to 10 objects. It checks object hashes and SQLite integrity, then replaces the index atomically. Upload the new index and restart/redeploy the app to activate a new snapshot; indexing is not automatic yet.
+
+The benchmark fetches one row group per source, validates its selected probability arrays, and compares first-load and cached retrieval plus plot aggregation. It writes `data/benchmark.json` with workspace measurements; these are not a promise of Space latency. The two large source groups require substantial first-load transfers even for a single selected segment. The full collection is never downloaded or indexed at app startup.
 
 ## Lookup behavior
 
 - Exact versioned accessions match only that version; unknown versions do not fall back.
-- Unversioned accessions return all matching versions in the sample. Input ignores case and surrounding whitespace.
-- Assembly lookups return all sampled contigs/segments for that assembly. This sample is not a complete assembly.
-- A miss means “not in this sample,” not “absent from GenBank/the bucket.”
+- Unversioned accessions resolve matching versions in the index. Input ignores case and surrounding whitespace.
+- Assembly lookups return indexed contigs/segments for that assembly, subject to the result limit. Indexed coverage may be incomplete.
+- A miss means “not in this indexed subset,” not “absent from GenBank/the bucket.”
 - Split contigs stay as separate segments, preserving their source coordinates and segment counts.
 
 ## Data and coordinates
@@ -62,7 +96,7 @@ The default source object is about 31 MB; automatic downloads are capped at 100
 python prepare_sample.py --local-file /path/to/source.parquet
 ```
 
-Source rows are selected intact, without truncating their probabilities. The app never scans or downloads the full bucket at startup. To support arbitrary accessions across the full bucket later, build a separate metadata index mapping accession → bucket object → row group, then retrieve matching probability data on demand.
+Source rows are selected intact, without truncating their probabilities. This command rebuilds only the optional offline sample, independently of the remote SQLite index.
 
 ## Checks
 

app.py

@@ -1,8 +1,10 @@
 """A small Hugging Face Space for exploring genome annotations by accession."""
 from pathlib import Path
+from contextlib import closing
 import os
 import re
 import tempfile
+import time
 
 # Shared compute hosts can have /tmp/gradio owned by a different user.
 os.environ.setdefault("GRADIO_TEMP_DIR", str(Path(tempfile.gettempdir()) / f"genbank-explorer-{os.getuid()}"))
@@ -11,23 +13,29 @@ import gradio as gr
 import pyarrow.parquet as pq
 
 from catalog import Catalog
+from remote_catalog import RemoteCatalog, RemoteReadError
 
 
 def build_app(catalog=None):
-    catalog = catalog or Catalog()
-    all_ids = list(range(len(catalog.records)))
-    first = catalog.records[0]
+    if catalog is None:
+        catalog = Catalog() if os.environ.get("GENBANK_DATA_MODE") == "sample" else RemoteCatalog()
+    remote_mode = isinstance(catalog, RemoteCatalog)
+    all_ids = catalog.browse_ids()
+    first = catalog.records[all_ids[0]]
+    scope = "indexed subset" if remote_mode else "sample"
 
     def search(accession):
-        ids = catalog.lookup(accession)
+        began = time.perf_counter()
+        ids, total = catalog.find(accession)
+        elapsed = time.perf_counter() - began
         choices = [(f"{catalog.records[i]['record_name']} · {catalog.records[i]['assembly_accession']} · "
                     f"[{catalog.records[i]['segment_start_bp']:,}, {catalog.records[i]['segment_end_bp']:,})", str(i)) for i in ids]
         if not str(accession or "").strip():
             message = "Enter an assembly or contig accession. Try an example below."
         elif not ids:
-            message = "No match in this sample. This does not mean the accession is absent from the full bucket."
+            message = f"No match in this {scope}. This does not mean the accession is absent from the full bucket."
         else:
-            message = f"Found **{len(ids)} sampled segment(s)**. Assembly results include only contigs loaded in this prototype."
+            message = f"Found **{total:,} indexed segment(s)** in {elapsed * 1000:.1f} ms. Showing {len(ids):,}. Assembly coverage may be partial."
         return message, catalog.table(ids), gr.Dropdown(choices=choices, value=str(ids[0]) if ids else None), None
 
     def select_segment(index):
@@ -36,31 +44,43 @@ def build_app(catalog=None):
         record = catalog.records[int(index)]
         start = record["segment_start_bp"]
         end = record["segment_end_bp"]
-        plot, step = catalog.window(index, start, end)
-        return record, start, end, plot, plot_note(step), None
+        try:
+            table, stats = catalog.fetch(index)
+            plot, step = catalog.window(index, start, end, table=table)
+        except RemoteReadError as exc:
+            return record, start, end, None, str(exc), None
+        return record, start, end, plot, plot_note(step, stats), None
 
-    def plot_note(step):
+    def plot_note(step, stats):
+        origin = "local sample" if stats.get("local") else "cache" if stats["cache_hit"] else "bucket"
         return ("Coordinates are **0-based, end-exclusive**. "
                 + ("Each point is one base." if step == 1 else f"Each point is the mean of up to **{step:,} bases**; short peaks can be smoothed.")
-                + " Download the segment for the original per-base probabilities.")
+                + " Download the segment for the original per-base probabilities.\n\n"
+                + f"Loaded from **{origin}** in **{stats['seconds']:.2f} s**"
+                + (f" · {stats['bytes_read'] / 1_000_000:.2f} MB fetched." if origin == "bucket" else "."))
 
     def update_window(index, start, end):
         if index is None:
             raise gr.Error("Look up an accession and choose a segment first.")
         try:
-            plot, step = catalog.window(index, start, end)
+            table, stats = catalog.fetch(index)
+            plot, step = catalog.window(index, start, end, table=table)
         except (ValueError, TypeError, OverflowError) as exc:
             raise gr.Error(str(exc)) from exc
-        return plot, plot_note(step)
+        return plot, plot_note(step, stats)
 
     def export(index):
         if index is None:
             raise gr.Error("Choose a segment first.")
-        table = catalog.segment_table(index)
+        try:
+            table = catalog.segment_table(index)
+        except RemoteReadError as exc:
+            raise gr.Error(str(exc)) from exc
         assembly = re.sub(r"[^A-Za-z0-9._-]", "_", table["assembly_accession"][0].as_py())
         record = re.sub(r"[^A-Za-z0-9._-]", "_", table["record_name"][0].as_py())
-        start = table["segment_start_bp"][0].as_py()
-        end = table["segment_end_bp"][0].as_py()
+        metadata = catalog.records[int(index)]
+        start = metadata["segment_start_bp"]
+        end = metadata["segment_end_bp"]
         filename = f"{assembly}__{record}__{start}-{end}.parquet"
         target = Path(tempfile.mkdtemp(prefix="genbank-export-")) / filename
         pq.write_table(table, target, compression="zstd")
@@ -68,15 +88,21 @@ def build_app(catalog=None):
 
     with gr.Blocks(title="GenBank Annotation Explorer", delete_cache=(3600, 3600)) as demo:
         gr.Markdown("# GenBank Annotation Explorer\nExplore predicted coding regions by **assembly or contig accession**.")
-        gr.Markdown(f"**Prototype sample · {len(catalog.records):,} segments · "
+        gr.Markdown(f"**Prototype {scope} · {len(catalog.records):,} segments · "
                     f"{len(catalog.manifest['assemblies']):,} assemblies · {catalog.manifest['bases']:,} bases**\n\n"
                     "Source: [HuggingFaceBio/genbank-annotations](https://huggingface.co/buckets/HuggingFaceBio/genbank-annotations). "
-                    "These are model-predicted CDS probabilities, not curated gene features. Search covers the loaded sample only.")
+                    f"These are model-predicted CDS probabilities, not curated gene features. Search covers the {scope} only. "
+                    + (f"Annotations load on demand from **{len(catalog.manifest['sources'])} bucket files**. "
+                       f"Index updated {catalog.manifest['created_at'][:10]}." if remote_mode else ""))
         with gr.Row():
             accession = gr.Textbox(label="Accession ID", placeholder="Assembly (GCA_…) or contig accession", scale=5)
             search_button = gr.Button("Find annotations", variant="primary", scale=1)
-        gr.Examples(examples=[[first["assembly_accession"]], [first["record_name"]],
-                              [first["record_name"].rsplit(".", 1)[0]]], inputs=accession)
+        examples = [first["assembly_accession"], first["record_name"]]
+        if remote_mode:
+            with closing(catalog.connect()) as conn:
+                examples += [r[0] for r in conn.execute("SELECT record_name FROM segments WHERE id IN (SELECT min(id) FROM segments GROUP BY object_path)")]
+            examples += ["JBPJTW010000350.1"]
+        gr.Examples(examples=[[e] for e in dict.fromkeys(examples)], inputs=accession)
         status = gr.Markdown("Enter an accession or select an example. IDs are case-insensitive; version suffixes are optional.")
         results = gr.Dataframe(value=catalog.table([]), interactive=False, label="Matching segments")
         segment = gr.Dropdown(choices=[], label="Segment to explore", interactive=True)
@@ -92,9 +118,11 @@ def build_app(catalog=None):
         with gr.Row():
             download = gr.Button("Prepare segment download")
             file = gr.File(label="Original segment annotations (Parquet)", interactive=False)
-        with gr.Accordion("Accessions available in this sample", open=False):
+        with gr.Accordion("Browse indexed accessions and coverage", open=False):
+            gr.Markdown(f"Showing the first {len(all_ids):,} indexed segments. Search an accession to find other indexed records. "
+                        "An indexed file does not imply complete coverage of its assembly.")
             gr.Dataframe(value=catalog.table(all_ids), interactive=False)
-            gr.JSON(value=catalog.manifest, label="Sample provenance")
+            gr.JSON(value=catalog.manifest, label="Index provenance" if remote_mode else "Sample provenance")
         outputs = [status, results, segment, file]
         for event in (search_button.click, accession.submit):
             event(search, accession, outputs).then(select_segment, segment, [metadata, start, end, plot, note, file])

benchmark_remote.py

@@ -0,0 +1,54 @@
+"""Measure one first-load and cached retrieval per indexed source file."""
+from contextlib import closing
+from datetime import datetime, timezone
+import json
+import platform
+import time
+
+import numpy as np
+
+from catalog import ROOT, PROBS
+from remote_catalog import RemoteCatalog
+
+
+def main():
+    catalog = RemoteCatalog()
+    with closing(catalog.connect()) as conn:
+        ids = [r[0] for r in conn.execute("SELECT min(id) FROM segments GROUP BY object_path ORDER BY object_path")]
+    results = []
+    for index in ids:
+        record = catalog.records[index]
+        start = time.perf_counter()
+        found = catalog.lookup(record["record_name"])
+        lookup_ms = (time.perf_counter() - start) * 1000
+        assert index in found
+        table, cold = catalog.fetch(index)
+        # Confirm that actual returned probability arrays agree with indexed coordinates.
+        length = record["segment_end_bp"] - record["segment_start_bp"]
+        for column in PROBS:
+            values = table[column][0].values.to_numpy()
+            assert len(values) == length
+            assert np.isfinite(values).all()
+            assert ((values >= 0) & (values <= 1)).all()
+        cached, warm = catalog.fetch(index)
+        assert warm["cache_hit"] and warm["bytes_read"] == 0 and table.equals(cached)
+        start = time.perf_counter()
+        frame, step = catalog.window(index, table=cached)
+        plot_seconds = time.perf_counter() - start
+        assert len(frame) <= 2400
+        result = {"accession": record["record_name"], "assembly": record["assembly_accession"],
+                  "organism": record["organism_name"], "bases": length,
+                  "lookup_ms": lookup_ms, "first_load": cold, "cached_load": warm,
+                  "plot_seconds": plot_seconds, "plot_bin_bp": step}
+        results.append(result)
+        print(json.dumps(result), flush=True)
+        del table, cached, values
+    report = {"created_at": datetime.now(timezone.utc).isoformat(),
+              "environment": platform.system() + " / Python " + platform.python_version(),
+              "note": "Workspace measurements; not Space latency. Empty app cache before each source's first load; upstream caches uncontrolled. Bytes count returned file ranges, excluding HTTP overhead.",
+              "index_bytes": catalog.path.stat().st_size, "results": results}
+    (ROOT / "data" / "benchmark.json").write_text(json.dumps(report, indent=2) + "\n")
+
+
+if __name__ == "__main__":
+    main()

build_index.py

@@ -0,0 +1,110 @@
+"""Build a metadata-only SQLite snapshot from an explicit, bounded source list."""
+import argparse
+from datetime import datetime, timezone
+import json
+import os
+from pathlib import Path
+import sqlite3
+import tempfile
+import time
+
+from huggingface_hub import HfApi, HfFileSystem, get_token
+import pyarrow.parquet as pq
+
+from catalog import ROOT, PROBS, normalize, unversioned, segment_metadata
+from remote_catalog import BUCKET, MeasuredFile, source_info
+
+SOURCES = [
+    "annotations/fungi/s0000/s0000.chunk-00510.parquet",
+    "annotations/protozoa/s0003/s0003.chunk-00017.parquet",
+    "annotations/vertebrate_other/s0001/s0001.chunk-00018.parquet",
+]
+
+
+def build(output, sources):
+    if not 1 <= len(sources) <= 10 or len(set(sources)) != len(sources):
+        raise ValueError("Choose 1–10 distinct published Parquet objects.")
+    if any(not s.startswith("annotations/") or not s.endswith(".parquet") for s in sources):
+        raise ValueError("Only published annotation Parquet objects can be indexed.")
+    api, fs = HfApi(token=get_token()), HfFileSystem(token=get_token())
+    output = Path(output)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    fd, name = tempfile.mkstemp(suffix=".sqlite", dir=output.parent)
+    os.close(fd)
+    temporary = Path(name)
+    conn = sqlite3.connect(temporary)
+    try:
+        conn.executescript("""
+            CREATE TABLE segments (
+                id INTEGER PRIMARY KEY, assembly_accession TEXT NOT NULL,
+                record_name TEXT NOT NULL, segment_start_bp INTEGER NOT NULL,
+                segment_end_bp INTEGER NOT NULL, object_path TEXT NOT NULL,
+                object_hash TEXT NOT NULL, row_group INTEGER NOT NULL,
+                row_in_group INTEGER NOT NULL, metadata_json TEXT NOT NULL);
+            CREATE TABLE aliases (alias TEXT NOT NULL, segment_id INTEGER NOT NULL,
+                PRIMARY KEY (alias, segment_id)) WITHOUT ROWID;
+            CREATE TABLE metadata (key TEXT PRIMARY KEY, value TEXT NOT NULL);
+        """)
+        manifest = {"bucket_id": BUCKET, "created_at": datetime.now(timezone.utc).isoformat(),
+                    "scope": "Explicit subset of published objects; not the full bucket or complete assemblies",
+                    "sources": [], "rows": 0, "bases": 0, "assemblies": [], "divisions": []}
+        assemblies, divisions = set(), set()
+        for path in sources:
+            source = source_info(api, BUCKET, path)
+            if source is None:
+                raise ValueError(f"Missing source: {path}")
+            started = time.perf_counter()
+            with MeasuredFile(fs, f"buckets/{BUCKET}/{path}") as remote:
+                pf = pq.ParquetFile(remote)
+                columns = [c for c in pf.schema_arrow.names if c not in PROBS and c != "sequence"]
+                group_sizes = []
+                for group in range(pf.num_row_groups):
+                    group_sizes.append(pf.metadata.row_group(group).total_byte_size)
+                    rows = pf.read_row_group(group, columns=columns, use_threads=False).to_pylist()
+                    for row_in_group, record in enumerate(rows):
+                        record = segment_metadata(record)
+                        index = manifest["rows"]
+                        conn.execute("INSERT INTO segments VALUES (?,?,?,?,?,?,?,?,?,?)", (
+                            index, record["assembly_accession"], record["record_name"],
+                            record["segment_start_bp"], record["segment_end_bp"], path,
+                            source.xet_hash, group, row_in_group, json.dumps(record)))
+                        aliases = {normalize(record["source_key"])}
+                        for field in ("assembly_accession", "record_name"):
+                            exact = normalize(record[field])
+                            aliases.update([exact, unversioned(exact)])
+                        conn.executemany("INSERT INTO aliases VALUES (?,?)", [(a, index) for a in aliases if a])
+                        manifest["rows"] += 1
+                        manifest["bases"] += record["segment_end_bp"] - record["segment_start_bp"]
+                        assemblies.add(record["assembly_accession"])
+                        divisions.add(record["division"])
+                item = {"path": path, "xet_hash": source.xet_hash, "object_bytes": source.size,
+                        "rows": pf.metadata.num_rows, "row_groups": pf.num_row_groups,
+                        "max_uncompressed_row_group_bytes": max(group_sizes),
+                        "metadata_bytes_fetched": remote.bytes_read, "range_reads": remote.range_reads,
+                        "index_seconds": round(time.perf_counter() - started, 3)}
+            after = source_info(api, BUCKET, path)
+            if after is None or after.xet_hash != source.xet_hash:
+                raise ValueError(f"Source changed while indexing: {path}")
+            manifest["sources"].append(item)
+            print(json.dumps(item), flush=True)
+        manifest.update(assemblies=sorted(assemblies), divisions=sorted(divisions))
+        conn.execute("INSERT INTO metadata VALUES ('manifest',?)", (json.dumps(manifest),))
+        conn.commit()
+        if conn.execute("PRAGMA integrity_check").fetchone()[0] != "ok":
+            raise ValueError("SQLite integrity check failed")
+        conn.close()
+        os.replace(temporary, output)
+        print(json.dumps({"index_bytes": output.stat().st_size, "rows": manifest["rows"],
+                          "assemblies": len(assemblies), "bases": manifest["bases"]}), flush=True)
+        return manifest
+    finally:
+        conn.close()
+        temporary.unlink(missing_ok=True)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--source", action="append", help="Repeat to select up to 10 published objects")
+    parser.add_argument("--output", type=Path, default=ROOT / "data" / "catalog.sqlite")
+    args = parser.parse_args()
+    build(args.output, args.source or SOURCES)

catalog.py

@@ -22,6 +22,17 @@ def unversioned(value):
     return re.sub(r"\.\d+$", "", value)
 
 
+def segment_metadata(record):
+    """Older outputs store a whole aligned contig without segment columns."""
+    record = dict(record)
+    if "segment_start_bp" not in record and "segment_end_bp" not in record:
+        record.update(segment_start_bp=0, segment_end_bp=record["aligned_bp_length"],
+                      segment_bp_length=record["aligned_bp_length"], segment_index=0, segment_count=1)
+    if not 0 <= record["segment_start_bp"] < record["segment_end_bp"]:
+        raise ValueError("Invalid segment coordinates in annotation metadata.")
+    return record
+
+
 class Catalog:
     def __init__(self, directory=ROOT / "data"):
         directory = Path(directory)
@@ -54,6 +65,20 @@ class Catalog:
     def table(self, ids):
         return pd.DataFrame([self.records[i] for i in ids], columns=DISPLAY_COLUMNS)
 
+    def find(self, accession, limit=200):
+        ids = self.lookup(accession)
+        return ids[:limit], len(ids)
+
+    def browse_ids(self, limit=200):
+        return list(range(min(limit, len(self.records))))
+
+    def fetch(self, index):
+        import time
+        started = time.perf_counter()
+        table = self.segment_table(index)
+        return table, {"cache_hit": False, "bytes_read": 0, "range_reads": 0,
+                       "seconds": time.perf_counter() - started, "local": True}
+
     def segment_table(self, index):
         index = int(index)
         if not 0 <= index < len(self.records):
@@ -61,14 +86,15 @@ class Catalog:
         group, row = self.locations[index]
         return pq.ParquetFile(self.path).read_row_group(group).slice(row, 1)
 
-    def window(self, index, start=None, end=None, max_points=1200):
+    def window(self, index, start=None, end=None, max_points=1200, table=None):
         record = self.records[int(index)]
         lo, hi = record["segment_start_bp"], record["segment_end_bp"]
         start = lo if start is None else int(start)
         end = hi if end is None else int(end)
         if not lo <= start < end <= hi:
             raise ValueError(f"Enter a range within [{lo:,}, {hi:,}) with start < end.")
-        table = self.segment_table(index)
+        if table is None:
+            table = self.segment_table(index)
         width = end - start
         step = max(1, (width + max_points - 1) // max_points)
         positions = np.arange(start, end, step)

data/benchmark.json

@@ -0,0 +1,71 @@
+{
+  "created_at": "2026-09-30T13:29:32.577183+00:00",
+  "environment": "Linux / Python 3.10.16",
+  "note": "Workspace measurements; not Space latency. Empty app cache before each source's first load; upstream caches uncontrolled. Bytes count returned file ranges, excluding HTTP overhead.",
+  "index_bytes": 86016,
+  "results": [
+    {
+      "accession": "JBPJTW010000303.1",
+      "assembly": "GCA_055472895.1",
+      "organism": "Phakopsora pachyrhizi",
+      "bases": 66123,
+      "lookup_ms": 0.5395635962486267,
+      "first_load": {
+        "cache_hit": false,
+        "bytes_read": 9178665,
+        "range_reads": 22,
+        "seconds": 4.416150398552418
+      },
+      "cached_load": {
+        "cache_hit": true,
+        "bytes_read": 0,
+        "range_reads": 0,
+        "seconds": 0.0001310594379901886
+      },
+      "plot_seconds": 0.0011919979006052017,
+      "plot_bin_bp": 56
+    },
+    {
+      "accession": "CM126017.1",
+      "assembly": "GCA_052324655.1",
+      "organism": "Trichomonas vaginalis",
+      "bases": 35626321,
+      "lookup_ms": 0.39654597640037537,
+      "first_load": {
+        "cache_hit": false,
+        "bytes_read": 185297497,
+        "range_reads": 17,
+        "seconds": 8.835233750753105
+      },
+      "cached_load": {
+        "cache_hit": true,
+        "bytes_read": 0,
+        "range_reads": 0,
+        "seconds": 0.00029289815574884415
+      },
+      "plot_seconds": 1.9516547014936805,
+      "plot_bin_bp": 29689
+    },
+    {
+      "accession": "CM057449.1",
+      "assembly": "GCA_016271365.2",
+      "organism": "Neoceratodus forsteri",
+      "bases": 84713947,
+      "lookup_ms": 0.3896690905094147,
+      "first_load": {
+        "cache_hit": false,
+        "bytes_read": 186392742,
+        "range_reads": 22,
+        "seconds": 7.675006306730211
+      },
+      "cached_load": {
+        "cache_hit": true,
+        "bytes_read": 0,
+        "range_reads": 0,
+        "seconds": 0.00034056883305311203
+      },
+      "plot_seconds": 2.610140291042626,
+      "plot_bin_bp": 70595
+    }
+  ]
+}

remote_catalog.py

@@ -0,0 +1,154 @@
+"""SQLite metadata lookup with measured, bounded reads from the annotation bucket."""
+from collections import OrderedDict
+from contextlib import closing
+from functools import lru_cache
+import json
+from pathlib import Path
+import sqlite3
+import threading
+import time
+
+from huggingface_hub import HfApi, HfFileSystem, get_token
+from huggingface_hub.hf_file_system import HfFileSystemFile
+import pyarrow.parquet as pq
+
+from catalog import Catalog, ROOT, PROBS, normalize, segment_metadata
+
+BUCKET = "HuggingFaceBio/genbank-annotations"
+
+
+def source_info(api, bucket, path):
+    return next((e for e in api.list_bucket_tree(bucket, path, recursive=False)
+                 if e.path == path and e.type == "file"), None)
+
+
+class MeasuredFile(HfFileSystemFile):
+    """Count returned range bytes, excluding HTTP headers and retries."""
+    def __init__(self, fs, path):
+        self.bytes_read = 0
+        self.range_reads = 0
+        super().__init__(fs, path, mode="rb", block_size=1024 * 1024, cache_type="none")
+
+    def _fetch_range(self, start, end):
+        data = super()._fetch_range(start, end)
+        self.bytes_read += len(data)
+        self.range_reads += 1
+        return data
+
+
+class RemoteReadError(ValueError):
+    pass
+
+
+class Records:
+    def __init__(self, catalog):
+        self.catalog = catalog
+
+    def __len__(self):
+        return self.catalog.manifest["rows"]
+
+    def __getitem__(self, index):
+        return self.catalog.record(int(index))
+
+
+class RemoteCatalog(Catalog):
+    def __init__(self, path=ROOT / "data" / "catalog.sqlite", cache_bytes=512 * 1024**2,
+                 max_group_bytes=512 * 1024**2, api=None, fs=None):
+        self.path = Path(path)
+        self.cache_limit = cache_bytes
+        self.max_group_bytes = max_group_bytes
+        self.api = api or HfApi(token=get_token())
+        self.fs = fs or HfFileSystem(token=get_token())
+        self.cache = OrderedDict()
+        self.cache_bytes = 0
+        self.lock = threading.Lock()
+        with closing(self.connect()) as conn:
+            self.manifest = json.loads(conn.execute("SELECT value FROM metadata WHERE key='manifest'").fetchone()[0])
+        self.records = Records(self)
+
+    def connect(self):
+        conn = sqlite3.connect(self.path.resolve().as_uri() + "?mode=ro&immutable=1", uri=True)
+        conn.row_factory = sqlite3.Row
+        return conn
+
+    @lru_cache(maxsize=512)
+    def entry(self, index):
+        with closing(self.connect()) as conn:
+            row = conn.execute("SELECT * FROM segments WHERE id=?", (int(index),)).fetchone()
+        if row is None:
+            raise RemoteReadError("Choose a segment from the current index.")
+        return dict(row)
+
+    def record(self, index):
+        return json.loads(self.entry(index)["metadata_json"])
+
+    def browse_ids(self, limit=200):
+        with closing(self.connect()) as conn:
+            return [r[0] for r in conn.execute("SELECT id FROM segments ORDER BY id LIMIT ?", (limit,))]
+
+    def find(self, accession, limit=200):
+        key = normalize(accession)
+        if not key:
+            return [], 0
+        # Aliases include exact versions and versionless IDs; never strip a query's version.
+        with closing(self.connect()) as conn:
+            count = conn.execute("SELECT count(*) FROM aliases WHERE alias=?", (key,)).fetchone()[0]
+            ids = [r[0] for r in conn.execute(
+                "SELECT s.id FROM aliases a JOIN segments s ON s.id=a.segment_id "
+                "WHERE a.alias=? ORDER BY s.assembly_accession,s.record_name,s.segment_start_bp,s.id LIMIT ?",
+                (key, limit))]
+        return ids, count
+
+    def lookup(self, accession):
+        return self.find(accession)[0]
+
+    def check_source(self, entry):
+        source = source_info(self.api, self.manifest["bucket_id"], entry["object_path"])
+        if source is None or source.xet_hash != entry["object_hash"]:
+            raise RemoteReadError("The bucket object has changed since indexing. Rebuild the index before retrieving this segment.")
+
+    def fetch(self, index):
+        began = time.perf_counter()
+        entry = self.entry(int(index))
+        key = (entry["object_path"], entry["object_hash"], entry["row_group"])
+        stats = {"cache_hit": False, "bytes_read": 0, "range_reads": 0}
+        # Serialize cache fills to bound memory and avoid duplicate remote downloads.
+        with self.lock:
+            if key in self.cache:
+                table = self.cache.pop(key)
+                self.cache[key] = table
+                stats["cache_hit"] = True
+            else:
+                try:
+                    self.check_source(entry)
+                    path = f"buckets/{self.manifest['bucket_id']}/{entry['object_path']}"
+                    self.fs.invalidate_cache(path)
+                    with MeasuredFile(self.fs, path) as remote:
+                        parquet = pq.ParquetFile(remote)
+                        group = parquet.metadata.row_group(entry["row_group"])
+                        if group.total_byte_size > self.max_group_bytes:
+                            raise RemoteReadError("This row group exceeds the prototype's 512 MiB read limit. It needs a smaller storage chunk.")
+                        table = parquet.read_row_group(entry["row_group"], use_threads=False)
+                        stats.update(bytes_read=remote.bytes_read, range_reads=remote.range_reads)
+                    self.check_source(entry)
+                except RemoteReadError:
+                    raise
+                except Exception as exc:
+                    raise RemoteReadError("Could not retrieve annotations from the bucket. Check bucket access and try again; this is not an accession-not-found result.") from exc
+                if table.nbytes <= self.cache_limit:
+                    while self.cache and self.cache_bytes + table.nbytes > self.cache_limit:
+                        _, old = self.cache.popitem(last=False)
+                        self.cache_bytes -= old.nbytes
+                    self.cache[key] = table
+                    self.cache_bytes += table.nbytes
+            result = table.slice(entry["row_in_group"], 1)
+            if result.num_rows != 1:
+                raise RemoteReadError("The retrieved segment does not match the index. Rebuild the index.")
+            actual = segment_metadata(result.select([c for c in result.column_names if c not in PROBS and c != "sequence"]).to_pylist()[0])
+            if any(actual[k] != entry[k] for k in ("record_name", "assembly_accession", "segment_start_bp", "segment_end_bp")):
+                raise RemoteReadError("The retrieved segment does not match the index. Rebuild the index.")
+        stats["seconds"] = time.perf_counter() - began
+        return result, stats
+
+    def segment_table(self, index):
+        return self.fetch(index)[0]

requirements.txt

@@ -2,3 +2,5 @@ gradio==6.6.0
 numpy>=1.26,<3
 pandas>=2.2,<3
 pyarrow>=18,<24
+huggingface_hub==1.16.0
+fsspec>=2024.6,<2027

tests/test_remote_catalog.py

@@ -0,0 +1,133 @@
+import io
+import json
+from pathlib import Path
+import sqlite3
+import tempfile
+from types import SimpleNamespace
+import unittest
+from unittest.mock import Mock, patch
+
+import pyarrow as pa
+import pyarrow.parquet as pq
+
+from remote_catalog import RemoteCatalog, RemoteReadError
+from catalog import segment_metadata
+
+
+class CountingBuffer(io.BytesIO):
+    def __init__(self, data):
+        super().__init__(data)
+        self.bytes_read = self.range_reads = 0
+
+    def read(self, n=-1):
+        data = super().read(n)
+        self.bytes_read += len(data)
+        self.range_reads += 1
+        return data
+
+
+class RemoteCatalogTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.path = Path(self.temp.name) / 'catalog.sqlite'
+        records = []
+        for i in range(3):
+            version = 1 if i < 2 else 2
+            start = 100 if i == 1 else 0
+            records.append(dict(assembly_accession=f'GCA_123.{version}', record_name=f'ABC123.{version}',
+                                source_key=f'GCA_123.{version}|ABC123.{version}', organism_name='Test',
+                                division='fungi', segment_index=int(i == 1), segment_count=2 if i < 2 else 1,
+                                segment_start_bp=start, segment_end_bp=start + 5,
+                                pred_prob_positive_strand_cds=[0., .2, .4, .6, .8],
+                                pred_prob_negative_strand_cds=[1., .8, .6, .4, .2]))
+        self.table = pa.Table.from_pylist(records)
+        sink = pa.BufferOutputStream()
+        pq.write_table(self.table, sink, row_group_size=1)
+        self.parquet = sink.getvalue().to_pybytes()
+        with sqlite3.connect(self.path) as conn:
+            conn.executescript('''
+                CREATE TABLE segments(id INTEGER PRIMARY KEY, assembly_accession TEXT, record_name TEXT,
+                    segment_start_bp INTEGER, segment_end_bp INTEGER, object_path TEXT, object_hash TEXT,
+                    row_group INTEGER, row_in_group INTEGER, metadata_json TEXT);
+                CREATE TABLE aliases(alias TEXT, segment_id INTEGER, PRIMARY KEY(alias,segment_id));
+                CREATE TABLE metadata(key TEXT PRIMARY KEY,value TEXT);
+            ''')
+            for i, record in enumerate(records):
+                metadata = {k: v for k, v in record.items() if not k.startswith('pred_prob_')}
+                conn.execute('INSERT INTO segments VALUES(?,?,?,?,?,?,?,?,?,?)', (
+                    i, record['assembly_accession'], record['record_name'], record['segment_start_bp'],
+                    record['segment_end_bp'], 'annotations/test.parquet', 'original', i, 0, json.dumps(metadata)))
+                aliases = {record['source_key'], 'GCA_123', 'ABC123', record['assembly_accession'], record['record_name']}
+                conn.executemany('INSERT INTO aliases VALUES(?,?)', [(a, i) for a in aliases])
+            conn.execute('INSERT INTO metadata VALUES(?,?)', ('manifest', json.dumps({'rows': 3, 'bucket_id': 'test/bucket'})))
+        self.api, self.fs = Mock(), Mock()
+        self.source = patch('remote_catalog.source_info', return_value=SimpleNamespace(xet_hash='original'))
+        self.source_mock = self.source.start()
+        self.reader = patch('remote_catalog.MeasuredFile', side_effect=lambda *args: CountingBuffer(self.parquet))
+        self.reader_mock = self.reader.start()
+        self.catalog = RemoteCatalog(self.path, api=self.api, fs=self.fs)
+
+    def tearDown(self):
+        self.reader.stop()
+        self.source.stop()
+        self.temp.cleanup()
+
+    def test_lookup_versions_segments_and_limit(self):
+        self.assertEqual(self.catalog.find(' abc123.1 '), ([0, 1], 2))
+        self.assertEqual(self.catalog.find('GCA_123', limit=2), ([0, 1], 3))
+        self.assertEqual(self.catalog.lookup('ABC123.2'), [2])
+        self.assertEqual(self.catalog.lookup('ABC123.9'), [])
+        self.assertEqual(self.catalog.lookup("' OR 1=1 --"), [])
+        self.assertEqual(self.catalog.lookup('GCA_123.1|ABC123.1'), [0, 1])
+
+    def test_cached_read_and_absolute_window(self):
+        table, cold = self.catalog.fetch(1)
+        self.assertFalse(cold['cache_hit'])
+        self.assertGreater(cold['bytes_read'], 0)
+        cached, warm = self.catalog.fetch(1)
+        self.assertTrue(warm['cache_hit'])
+        self.assertEqual(warm['bytes_read'], 0)
+        self.assertTrue(table.equals(cached))
+        self.assertEqual(self.reader_mock.call_count, 1)
+        frame, step = self.catalog.window(1, table=table)
+        self.assertEqual(frame['Position (bp)'].min(), 100)
+        self.assertEqual(frame['Position (bp)'].max(), 104)
+        self.assertEqual(self.reader_mock.call_count, 1)
+
+    def test_cache_budget_and_eviction(self):
+        self.catalog.cache_limit = pq.ParquetFile(io.BytesIO(self.parquet)).read_row_group(0).nbytes
+        self.catalog.fetch(0)
+        self.catalog.fetch(1)
+        self.assertLessEqual(self.catalog.cache_bytes, self.catalog.cache_limit)
+        self.assertEqual(len(self.catalog.cache), 1)
+        _, stats = self.catalog.fetch(0)
+        self.assertFalse(stats['cache_hit'])
+
+    def test_stale_source_rejected_before_and_after_read(self):
+        self.source_mock.return_value = SimpleNamespace(xet_hash='changed')
+        with self.assertRaisesRegex(RemoteReadError, 'changed'):
+            self.catalog.fetch(0)
+        self.assertEqual(self.reader_mock.call_count, 0)
+        self.source_mock.side_effect = [SimpleNamespace(xet_hash='original'), SimpleNamespace(xet_hash='changed')]
+        with self.assertRaisesRegex(RemoteReadError, 'changed'):
+            self.catalog.fetch(0)
+        self.assertEqual(len(self.catalog.cache), 0)
+
+    def test_oversized_group_and_invalid_id(self):
+        self.catalog.max_group_bytes = 1
+        with self.assertRaisesRegex(RemoteReadError, 'read limit'):
+            self.catalog.fetch(0)
+        with self.assertRaises(RemoteReadError):
+            self.catalog.fetch(-1)
+
+    def test_auth_failure_is_not_a_lookup_miss(self):
+        self.source_mock.side_effect = RuntimeError('sensitive upstream detail')
+        with self.assertRaisesRegex(RemoteReadError, 'Could not retrieve') as error:
+            self.catalog.fetch(0)
+        self.assertNotIn('sensitive', str(error.exception))
+        self.assertEqual(self.catalog.lookup('ABC123.2'), [2])
+
+    def test_legacy_contig_coordinates(self):
+        record = segment_metadata({'aligned_bp_length': 35000, 'record_name': 'OLD.1'})
+        self.assertEqual((record['segment_start_bp'], record['segment_end_bp']), (0, 35000))
+        self.assertEqual((record['segment_index'], record['segment_count']), (0, 1))