Indexing Catalog Collections for Agent Retrieval

A collection manifest says a dataset covers a country between 2017 and now. Its items say which square kilometres were actually captured on which days, and the gap between those two statements is where agents produce confident answers about places nobody photographed. This guide builds the two-level index that closes it, implementing the catalog stage of spatial metadata and catalog indexing.

When to Use This Approach

Index at both levels whenever a collection’s coverage is uneven, which is nearly always for anything captured rather than modelled. A single-level index is adequate only for products that genuinely blanket their declared extent.

Collection type	Coverage	Index
Modelled national product	Complete by construction	Collection level only
Satellite archive	Dense but cloud-gapped	Both levels
Aerial survey programme	Patchy, campaign-based	Both levels, item extent essential
Ground survey records	Sparse points	Item level, collection extent is decoration
Derived analytic layer	Follows its input	Both, and record the lineage

The distinction that matters is whether “covers” means “is defined over” or “was observed across”. A modelled layer is defined everywhere in its extent; an archive was observed wherever the sensor happened to look.

The declared extent is a bounding rectangle around observations, not a statement about them. Treating it as coverage is what lets an agent answer confidently about a region the sensor never saw.

Implementation

The ingester walks collections, normalises each into the catalog record shape, then pages through items and stores their footprints in a separate table keyed back to the collection.

import logging
from dataclasses import dataclass
from datetime import date
from typing import Iterable, Iterator, Optional

log = logging.getLogger("catalog_ingest")

MAX_ITEM_PAGES = 500          # a guard against an endless paging loop


@dataclass(frozen=True)
class Item:
    item_id: str
    collection_id: str
    bbox: tuple[float, float, float, float]
    captured: Optional[date]
    usable_fraction: Optional[float]     # e.g. cloud-free share, when the source reports it


def _paged_items(client, collection_id: str, page_size: int = 500) -> Iterator[dict]:
    """Yield raw item records, stopping on error rather than aborting the whole ingest."""
    token, pages = None, 0
    while pages < MAX_ITEM_PAGES:
        try:
            batch, token = client.items(collection_id, limit=page_size, token=token)
        except Exception as exc:
            log.warning("item paging failed for %s after %d page(s): %s",
                        collection_id, pages, exc)
            return                                  # partial coverage beats no ingest
        if not batch:
            return
        yield from batch
        pages += 1
        if not token:
            return
    log.warning("item paging for %s hit the page guard at %d", collection_id, MAX_ITEM_PAGES)


def normalise_item(raw: dict, collection_id: str) -> Optional[Item]:
    """Return an item, or None when it cannot supply a usable footprint."""
    try:
        bbox = tuple(float(v) for v in raw["bbox"][:4])
        if bbox[0] > bbox[2] or bbox[1] > bbox[3]:
            log.warning("item %s has a degenerate footprint", raw.get("id"))
            return None
        return Item(
            item_id=str(raw["id"]),
            collection_id=collection_id,
            bbox=bbox,
            captured=_as_date(raw.get("properties", {}).get("datetime")),
            usable_fraction=_usable_fraction(raw.get("properties", {})),
        )
    except (KeyError, TypeError, ValueError) as exc:
        log.warning("rejecting malformed item %r: %s", raw.get("id"), exc)
        return None


def ingest_collection(client, collection_id: str, upsert_item) -> dict:
    """Ingest one collection's items; report counts so partial ingests are visible."""
    stored = rejected = 0
    for raw in _paged_items(client, collection_id):
        item = normalise_item(raw, collection_id)
        if item is None:
            rejected += 1
            continue
        upsert_item(item)
        stored += 1
    stats = {"collection": collection_id, "stored": stored, "rejected": rejected}
    log.info("catalog item ingest: %s", stats)
    return stats

Two decisions here are about failing well. Paging stops on error rather than raising, because a collection with fifty thousand items and a flaky endpoint should contribute the forty thousand it managed rather than nothing — provided the partial state is recorded, which the counts do. And the page guard exists because a paging token that never advances is a real failure mode of paginated APIs, and without a bound it becomes an infinite loop in a nightly job.

The item table wants a spatial index and a date index, and the queries against it are the same index-aware shape used everywhere in this section:

CREATE TABLE catalog_items (
    item_id       text PRIMARY KEY,
    collection_id text NOT NULL REFERENCES catalog_collections(collection_id),
    geom          geometry(Polygon, 4326) NOT NULL,
    captured      date,
    usable_frac   double precision
);

CREATE INDEX catalog_items_geom_idx ON catalog_items USING gist (geom);
CREATE INDEX catalog_items_when_idx ON catalog_items (collection_id, captured DESC);

-- Does this collection actually observe this place, recently enough, usably enough?
SELECT count(*) AS observations,
       max(captured) AS most_recent
FROM   catalog_items
WHERE  collection_id = :collection
  AND  geom && :region                     -- index-aware pre-filter
  AND  ST_Intersects(geom, :region)
  AND  captured >= :since
  AND  coalesce(usable_frac, 1.0) >= :min_usable;

Two very different spatial distributions. Mixing them in one table gives the index a bimodal workload it serves badly for both; separating them lets each index specialise, at the cost of one extra join nobody misses.

Validation & Testing

def test_items_fall_inside_their_collection_extent(conn):
    stray = conn.execute("""
        SELECT i.item_id FROM catalog_items i
        JOIN catalog_collections c USING (collection_id)
        WHERE NOT ST_Within(i.geom, ST_Expand(c.geom, 0.01))
        LIMIT 5""").fetchall()
    assert not stray, f"items outside their collection extent: {stray}"


def test_partial_ingest_is_reported_not_silent(caplog):
    stats = ingest_collection(FlakyClient(fail_after=3), "col-a", upsert)
    assert stats["stored"] > 0
    assert any("item paging failed" in r.message for r in caplog.records)


def test_paging_guard_terminates_on_a_stuck_token():
    stats = ingest_collection(StuckTokenClient(), "col-b", upsert)
    assert stats["stored"] <= MAX_ITEM_PAGES * 500


def test_degenerate_item_footprint_is_rejected():
    assert normalise_item({"id": "x", "bbox": [10, 10, 5, 5]}, "col-a") is None

Run the first test as a report rather than only as a gate when a catalog is first ingested: a handful of stray items is a data-quality note worth sending upstream, while a third of a collection sitting outside its declared extent means the extent field is being populated by something other than the items and should not be trusted for shortlisting at all.

The first test is the one that catches upstream data problems rather than code problems, and it will occasionally fail on legitimately odd data — a collection whose declared extent was never updated after items were added outside it. That is worth knowing about, which is why it asserts rather than warns.

Gotchas & Edge Cases

Item footprints that are the whole scene, not the usable part. A satellite scene’s footprint covers the full swath including the cloudy half. Store the usable fraction where the source reports it and require a threshold in coverage queries, or coverage counts overstate what is actually visible.

Collections with millions of items and no need for them. Ingesting every item of a global archive to answer questions about one country is a great deal of storage for no benefit. Restrict item ingest to your regions of interest and record that restriction, so a later query outside them reports “not indexed” rather than “not covered”.

A footprint is where the sensor pointed, not what it saw. Coverage counts that ignore the usable fraction are systematically optimistic in exactly the regions — persistently cloudy ones — where the question is worth asking.

Items with identifiers that are not stable across reprocessing. Some archives reissue items with new identifiers after reprocessing, which duplicates coverage in the index. Deduplicate on the tuple of collection, footprint and capture date rather than trusting identifiers alone.

Date fields that are ranges, not instants. Composite products carry a start and an end, and storing only one of them makes temporal queries wrong at the boundaries. Store both, or store the range type your database offers.

Foreign keys that block ingest ordering. Items referencing a collection that has not been ingested yet will fail insertion. Ingest collections first in the same transaction, or stage items and resolve references afterwards; discovering this in production means a half-ingested catalog.

Frequently Asked Questions

How fresh does the item index need to be?

As fresh as the questions being asked. A corpus supporting historical analysis can refresh weekly; one answering "what does the latest imagery show" needs to be within a day of the archive. Publish the index age alongside answers so a stale index degrades gracefully into a dated answer rather than a wrong one — an agent that says "as of last Tuesday" is far more useful than one that implies currency it does not have.

Should item geometry be the true footprint or its bounding box?

The true footprint when the source provides it, because scene footprints are frequently rotated quadrilaterals whose bounding box overstates coverage by a third. That overstatement lands exactly where it hurts: at the edges of a coverage gap, where the question of whether a place was observed is genuinely in doubt.

What should a coverage query return when nothing is indexed?

A distinguishable "not indexed" rather than a zero count. Those two states mean opposite things — one says the place was never observed, the other says nobody asked the archive about it — and collapsing them into a zero produces an agent that confidently reports absence of data it simply never fetched.

Is it worth storing item-level metadata beyond footprint and date?

Only the fields a query will filter on: usable fraction, processing level, sensor mode. Everything else can be fetched from the source when an item is actually selected, which is rare compared to how often items are counted. Storing the full item record for millions of items inflates the index by an order of magnitude to serve a lookup that happens once per answer.

Finally, record the ingest restrictions somewhere the query layer can read. An index that deliberately holds items for three regions is a correct design and a dangerous one if the query layer does not know the boundary: outside it, every coverage question returns zero, and zero is indistinguishable from a genuine gap unless the restriction is data rather than folklore.

Up to the parent topic: Spatial Metadata and Catalog Indexing
Mapping Catalog Fields to Retrieval Filters
Concept: Vector-Raster Hybrid Processing
Technique: Filtering Retrieval by Bounding Box Before Vector Search