# Bounding Cross-Domain Join Cost with Query Budgets

A federated join whose cost is discovered by running it will eventually be the query that takes down the estate. Budgets move that discovery to admission time: the query declares an extent, the planner estimates what it will scan, and anything above the ceiling is refused with a diagnostic instead of attempted with hope. This guide implements admission, enforcement during execution, and the refusal semantics that send a consumer to the right path rather than merely away. It operationalises the budget specification in Federated Query and Cross-Domain Spatial Joins within Federated Ownership & Routing Architecture, and refused queries land on the path described in Async Execution for Heavy Spatial Queries.

# Prerequisites

# Step-by-Step Implementation

# 1. Refuse the queries that have no bounded form at admission

Some queries cannot be budgeted because their cost is not a function of anything supplied. Those are refused before estimation, and the refusal names the async path.

# admission.py — the checks that precede any estimation.
UNBOUNDED_PREDICATES = {"ST_Union", "ST_ClusterDBSCAN", "ST_ClusterKMeans"}


class Refused(Exception):
    """Carries the reason and the alternative, so a consumer knows what to do next."""

    def __init__(self, reason: str, alternative: str):
        super().__init__(f"{reason} — {alternative}")
        self.reason, self.alternative = reason, alternative


def admit(query: dict) -> None:
    if not query.get("extent"):
        raise Refused(
            "no extent supplied; an unbounded join has no bounded cost",
            "supply an extent, or submit to the async path",
        )
    if not query.get("crs"):
        raise Refused(
            "no join CRS declared; mismatched SRIDs return an empty result silently",
            "declare crs explicitly on the request",
        )
    used = set(query.get("functions", []))
    if used & UNBOUNDED_PREDICATES:
        raise Refused(
            f"{sorted(used & UNBOUNDED_PREDICATES)} has no bounded form over two domains",
            "submit to the async path with a job budget",
        )
    if query.get("nearest_neighbour") and not query.get("max_distance_m"):
        raise Refused(
            "unbounded nearest-neighbour scans both sides entirely",
            "supply max_distance_m, or submit to the async path",
        )

Verify each refusal names an alternative — a refusal without one converts into a support conversation:

python3 - <<'PY'
import admission
for q in [{}, {"extent": "POLYGON(...)"},
          {"extent": "P", "crs": "EPSG:4326", "functions": ["ST_Union"]}]:
    try:
        admission.admit(q)
    except admission.Refused as e:
        print(f"{e.reason[:44]:46} -> {e.alternative}")
PY

# 2. Estimate the scan from catalog statistics before planning

The estimate does not need to be accurate; it needs to be conservative and cheap. Feature density over the requested extent gets there.

# estimate.py — conservative pre-flight cost, from catalog statistics only.
import math


def area_km2(extent_bbox: tuple[float, float, float, float]) -> float:
    """Approximate area of a lat/lon bbox. Conservative: overestimates near poles,
    which is the safe direction for an admission check."""
    minx, miny, maxx, maxy = extent_bbox
    mid_lat = math.radians((miny + maxy) / 2)
    km_per_deg_lat = 110.574
    km_per_deg_lon = 111.320 * math.cos(mid_lat)
    return abs(maxx - minx) * km_per_deg_lon * abs(maxy - miny) * km_per_deg_lat


def estimate_candidates(product_stats: dict, extent_bbox) -> int:
    """Rows a side is expected to contribute. Density is published by each product
    as features per square kilometre, refreshed on every publish."""
    return int(product_stats["feature_density_per_km2"] * area_km2(extent_bbox))


def estimate_pairs(left: dict, right: dict, extent_bbox) -> dict:
    l = estimate_candidates(left, extent_bbox)
    r = estimate_candidates(right, extent_bbox)
    return {
        "left_candidates": l,
        "right_candidates": r,
        # Envelope pairing is the dominant cost before the exact predicate runs.
        "envelope_pairs": l * r,
        "estimated_bytes": (l * left["avg_bytes"]) + (r * right["avg_bytes"]),
    }

Verify the estimate against a handful of real queries — it should over-predict, never under-predict:

python3 - <<'PY'
import estimate, json
left  = {"feature_density_per_km2": 41.2, "avg_bytes": 780}
right = {"feature_density_per_km2": 3.1,  "avg_bytes": 2100}
print(json.dumps(estimate.estimate_pairs(left, right, (-1, 50, 1, 52)), indent=2))
PY

# 3. Enforce the ceiling and refuse with the observed value

A refusal that says “too expensive” teaches nothing. One that says what was estimated, against which limit, lets a consumer narrow their extent themselves.

# enforce.py — the ceiling, and a refusal a consumer can act on.
CEILINGS = {
    "max_candidates_per_side": 250_000,
    "max_envelope_pairs": 20_000_000_000,
    "max_bytes_transferred": 256 * 1024 * 1024,
}


def enforce(est: dict) -> None:
    for side in ("left_candidates", "right_candidates"):
        if est[side] > CEILINGS["max_candidates_per_side"]:
            raise Refused(
                f"{side}={est[side]:,} exceeds {CEILINGS['max_candidates_per_side']:,}",
                f"narrow the extent by about {est[side] / CEILINGS['max_candidates_per_side']:.1f}×",
            )
    if est["envelope_pairs"] > CEILINGS["max_envelope_pairs"]:
        raise Refused(
            f"envelope_pairs={est['envelope_pairs']:,.0f}",
            "narrow the extent, or submit to the async path",
        )
    if est["estimated_bytes"] > CEILINGS["max_bytes_transferred"]:
        raise Refused(
            f"estimated transfer {est['estimated_bytes'] / 1e6:.0f}MB",
            "predicate pushdown may not be engaging — check the plan",
        )

# 4. Enforce again during execution, because estimates are estimates

Pre-flight estimation prevents the obviously impossible; runtime enforcement catches the case where statistics were stale.

# runtime.py — a progress callback that aborts with an actionable diagnostic.
import time


def run_with_budget(execute, query: dict, ceilings: dict) -> dict:
    started = time.monotonic()
    seen = {"rows": 0, "bytes": 0}

    def progress(rows: int, byts: int) -> None:
        seen["rows"] += rows
        seen["bytes"] += byts
        elapsed = time.monotonic() - started
        if seen["rows"] > ceilings["max_rows_scanned"]:
            raise Refused(
                f"aborted after {seen['rows']:,} rows — the estimate was low",
                "statistics may be stale; re-publish density, then retry",
            )
        if elapsed * 1000 > ceilings["max_duration_ms"]:
            raise Refused(
                f"aborted after {elapsed:.0f}s",
                "submit to the async path — this shape does not fit a request",
            )

    rows, provenance = execute(query, progress)
    return {"rows": rows, "provenance": provenance,
            "budget": {**seen, "duration_s": round(time.monotonic() - started, 1)}}

Verify an over-budget query aborts quickly rather than running to a timeout:

curl -sS -X POST "$ENGINE_URL/federated-query" -H 'content-type: application/json' \
  -d '{"extent":"POLYGON((-10 45,10 45,10 60,-10 60,-10 45))","crs":"EPSG:4326",
       "left":"cadastral/parcels","right":"hydrology/flood_extents"}' \
  | jq '{refused: .error, alternative: .alternative}'

# Configuration Reference

# Common Failure Modes & Fixes

Everything is refused after a bulk import. Root cause: feature density was republished and is now much higher, so estimates scale accordingly. Fix: this is the system working — consumers narrow extents. If it is genuinely wrong, the density statistic is being computed over the wrong extent.

Queries are admitted and then abort at runtime. Root cause: stale density statistics under-estimating. Fix: refresh density on every publish rather than on a schedule; the estimate is only as good as its input.

Consumers work around budgets by issuing many small queries. Root cause: entirely rational, and usually fine — many small bounded queries are cheaper than one large one. Fix: watch for pathological cases where the tiling overhead exceeds the saving, and offer those consumers the async path explicitly.

A refusal gives no usable guidance. Root cause: the message reports a limit without the observed value or a narrowing factor. Fix: the messages above; “narrow by about 3.4×” is actionable in a way “too large” is not.

A budget breach is really a pushdown failure. Root cause: bytes transferred is the tell — a query whose predicate is not pushing down transfers everything regardless of extent. Fix: check the plan before adjusting any ceiling.

# FAQ

# Related

• Federated Query and Cross-Domain Spatial Joins — the parent topic and the budget specification
• Async Execution for Heavy Spatial Queries — where refused queries belong
• Pushing Spatial Predicates Down with PostGIS FDW — why a bytes breach usually means pushdown failed