# Optimizing async execution for spatial joins

Cross-boundary polygon intersections and point-in-polygon validations across federated domains routinely exhaust synchronous connection pools and trigger cascading ingress timeouts. This guide is a concrete operation under Async Execution for Heavy Spatial Queries, itself part of the Federated Ownership & Routing Architecture: it walks through tuning a bounded work-queue pipeline that materializes spatial join results off the request path while preserving deterministic ordering and exactly-once delivery. The worker pool consumes only payloads that have already cleared Schema Contracts for Vector/Tile Data, so every step below assumes a validated, EPSG:4326 geometry contract at the edge.

# Prerequisites

# Step-by-Step Implementation

# 1. Provision a spatially-partitioned topic

Each partition maps to one domain’s bounding box so join fan-out stays localized and a single hot domain cannot starve its neighbours. Partition count is the upper bound on worker parallelism.

# One partition per spatial domain; durable acks for zero loss on failover.
kafka-topics --bootstrap-server "$BROKER_BOOTSTRAP" \
  --create --topic spatial-join-jobs \
  --partitions 12 --replication-factor 3 \
  --config retention.ms=3600000 --config min.insync.replicas=2

Verify the partition layout matches the domain registry before routing any jobs:

kafka-topics --bootstrap-server "$BROKER_BOOTSTRAP" \
  --describe --topic spatial-join-jobs | grep -E 'PartitionCount|Leader'

# 2. Configure the worker pool orchestrator

The worker manifest pins memory caps, concurrency, and the join template. acks: all with min.insync.replicas=2 guarantees no message loss during a broker election; linger_ms and batch_size raise throughput for geometry-heavy payloads.

async_spatial_join:
  broker:
    type: kafka
    partitions: 12
    retention_ms: 3600000
    acks: all
    linger_ms: 5
    batch_size: 16384
  worker_pool:
    concurrency: 8
    max_heap_mb: 4096
    spatial_engine: postgis
    query_template: |
      SELECT a.domain_id, b.feature_id, ST_Intersection(a.geom, b.geom) AS intersect_geom
      FROM domain_a_polygons a
      JOIN domain_b_points b ON ST_Intersects(a.geom, b.geom)
      WHERE a.spatial_partition = $1
      AND a.geom && ST_MakeEnvelope($2, $3, $4, $5, 4326)
      ORDER BY a.spatial_partition, a.geom
  materialization:
    strategy: parquet_s3
    compression: zstd
    schema_version: v2.1
    idempotency_key: trace_id
    retry_policy: exponential_backoff
    max_retries: 3

Confirm the pool rolled out and is consuming its partitions:

kubectl get deploy spatial-join-worker -n geospatial-mesh -o jsonpath='{.status.readyReplicas}'
kafka-consumer-groups --bootstrap-server "$BROKER_BOOTSTRAP" \
  --group spatial-join-workers --describe

# 3. Enforce bounding-box pre-filtering in the join

The && operator applies a GiST bounding-box test before the expensive ST_Intersection topology calculation, which keeps the planner off full-table scans across federated tables. Never ship a join template without it.

EXPLAIN (ANALYZE, BUFFERS, FORMAT JSON)
SELECT a.domain_id, b.feature_id, ST_Intersection(a.geom, b.geom) AS intersect_geom
FROM domain_a_polygons a
JOIN domain_b_points b ON ST_Intersects(a.geom, b.geom)
WHERE a.spatial_partition = 'us-east-1'
AND a.geom && ST_MakeEnvelope(-74.5, 40.1, -73.9, 40.9, 4326);

A healthy plan shows an Index Scan using idx_domain_a_geom. A Seq Scan means the GiST index is missing or ANALYZE statistics are stale — fix it before tuning anything else.

# 4. Guarantee exactly-once materialization

Bind every submission to a trace_id fingerprint and use a conditional write so a retried or rebalanced job can never double-insert. This mirrors the idempotency guarantees used in Cross-Domain Routing Strategies.

INSERT INTO spatial_join_results (trace_id, partition_key, result_blob, materialized_at)
VALUES ($1, $2, $3, NOW())
ON CONFLICT (trace_id, partition_key) DO NOTHING;

Verify that replaying the same trace_id yields exactly one row:

SELECT trace_id, COUNT(*) FROM spatial_join_results
GROUP BY trace_id HAVING COUNT(*) > 1;  -- must return zero rows

# 5. Route invalid payloads to a dead-letter queue

Materialized output adheres to schema_version: v2.1; any geometry that fails validation is diverted rather than corrupting downstream tile servers. Pre-filter heavy candidates with ST_DWithin before ST_Intersection to shrink the compute surface, and gate egress through API Gateway Mapping for GIS Services.

# Inspect what the worker quarantined and why.
kafka-console-consumer --bootstrap-server "$BROKER_BOOTSTRAP" \
  --topic spatial-join-jobs.DLQ --from-beginning --max-messages 20 | \
  jq '{trace_id, partition_key, reject_reason}'

# Configuration Reference

# Common Failure Modes & Fixes

• Symptom: join latency spikes above the 90s SLA. Root cause: Seq Scan on domain_a_polygons from a missing or fragmented GiST index. Fix: REINDEX INDEX CONCURRENTLY idx_domain_a_geom; during an off-peak window, then re-run ANALYZE domain_a_polygons;.
• Symptom: worker pods restart with OutOfMemoryError. Root cause: unbounded geometry serialization on a dense partition. Fix: lower concurrency, confirm max_heap_mb is enforced, and pre-filter with ST_DWithin before ST_Intersection.
• Symptom: consumer lag climbs past 10k messages. Root cause: broker backpressure or partition skew. Fix: add consumer instances and tune max.poll.records; rebalance hot partitions across the domain registry.
• Symptom: downstream tile server rejects results. Root cause: schema drift from a v2.0 producer writing a v2.1 topic. Fix: roll the producer back to the v2.0 contract, validate geometries with ST_IsValid, and replay from the DLQ.
• Symptom: index bloat degrades ST_Intersects. Root cause: pg_indexes_size exceeds 60% of pg_total_relation_size. Fix: run pg_repack (or VACUUM FULL in a maintenance window) to reclaim space without a long table lock.

# FAQ

# Related

• Async Execution for Heavy Spatial Queries — parent reference for the broker and worker-pool pattern
• Schema Contracts for Vector/Tile Data — the validation gate every payload must clear first
• API Gateway Mapping for GIS Services — synchronous-fallback routing for degraded requests