Aggregate regional bathymetry partitions

backend/README.md

@@ -1743,6 +1743,14 @@ manifest after every partition. Repeating it reuses completed source
 identities; `--force` deliberately refreshes them. A partial `--members` or
 `--max-partitions` run never marks regional coverage complete.
 
+Regional map analysis uses
+`POST /api/v1/projects/{project_id}/datasets/bathymetry/profiles/partitions/select`.
+It selects the latest complete manifest, prefilters overlapping municipality
+partitions and performs one PostGIS query over their persisted
+`vector_features`. Municipality Areas use only their exact partition. The
+response and server-side map export retain every contributing Dataset id and
+never substitute a single municipality Dataset for all of Flanders.
+
 Inspect the MDK North Sea WCS without downloading coverage:
 
 ```bash

backend/app/api/routes/datasets.py

@@ -257,6 +257,41 @@ def finalize_bathymetry_profile_partitions(
     return envelope(BathymetryProfileAcquisitionService.finalize_partitions(db, project_id, payload))
 
 
+@router.post("/datasets/bathymetry/profiles/partitions/select", response_model=dict)
+def select_bathymetry_profile_partitions(
+    project_id: UUID,
+    payload: VectorSelectionRequest,
+    db: Session = Depends(get_db),
+):
+    selection_geometry = None
+    selection_area_id = None
+    partition_area_id = None
+    if payload.area_id is not None:
+        selection_area = db.get(Area, payload.area_id)
+        if selection_area is None or selection_area.project_id != project_id:
+            raise AppError(code="AREA_NOT_FOUND", message="Area not found", status_code=404)
+        selection_geometry, _covers_full_area = VectorFeatureService.constrain_bbox_to_area(
+            payload.bbox.model_dump(),
+            selection_area.geometry,
+        )
+        selection_area_id = selection_area.id
+        if str(selection_area.name or "").lower().startswith("gemeente "):
+            partition_area_id = selection_area.id
+
+    result = VectorFeatureService.select_partitioned_features_by_bbox(
+        db,
+        project_id=project_id,
+        source_name=BathymetryProfileAcquisitionService.PROVIDER,
+        partition_scope_key="flanders",
+        bbox=payload.bbox.model_dump(),
+        limit=payload.limit,
+        selection_geometry=selection_geometry,
+        selection_area_id=selection_area_id,
+        partition_area_id=partition_area_id,
+    )
+    return envelope(VectorSelectionResponse(**result).model_dump(exclude_none=True))
+
+
 @router.post("/datasets/thematic-raster/acquire", response_model=dict)
 def acquire_bounded_thematic_raster(
     project_id: UUID,

backend/app/schemas/export.py

@@ -56,6 +56,7 @@ class MapResultExportRequest(BaseModel):
     area_id: UUID | None = None
     partitioned: bool = False
     product_key: str | None = None
+    partition_scope_key: str | None = None
     theme_id: str | None = None
     name: str | None = None
 
@@ -67,8 +68,8 @@ class MapResultExportRequest(BaseModel):
             self.earlier_dataset_id is None or self.later_dataset_id is None
         ):
             raise ValueError("earlier_dataset_id and later_dataset_id are required for evolution exports")
-        if self.partitioned and not self.product_key:
-            raise ValueError("product_key is required for partitioned raster exports")
+        if self.partitioned and not self.product_key and not self.partition_scope_key:
+            raise ValueError("product_key or partition_scope_key is required for partitioned exports")
         return self
 
 

backend/app/schemas/operations.py

@@ -251,3 +251,8 @@ class VectorSelectionResponse(BaseModel):
     truncated: bool
     geojson: dict
     summary: VectorSelectionSummary | None = None
+    partition_count: int | None = None
+    available_partition_count: int | None = None
+    partition_scope_key: str | None = None
+    source_name: str | None = None
+    dataset_ids: list[UUID] | None = None

backend/app/services/export_service.py

@@ -87,6 +87,16 @@ class ExportService:
         if not dataset or dataset.project_id != payload.project_id:
             raise AppError(code="DATASET_NOT_FOUND", message="Dataset not found", status_code=404)
         if dataset.dataset_type in DatasetService.VECTOR_TYPES:
+            if payload.partitioned:
+                return ExportService.export_partitioned_vector_selection_geojson(
+                    db,
+                    dataset,
+                    payload.bbox.model_dump(mode="json"),
+                    partition_scope_key=payload.partition_scope_key or "",
+                    area_id=payload.area_id,
+                    name=payload.name,
+                    limit=1000,
+                )
             return ExportService.export_vector_selection_geojson(
                 db,
                 dataset.id,
@@ -198,6 +208,87 @@ class ExportService:
         )
         return ExportService._create_response(export)
 
+    @staticmethod
+    def export_partitioned_vector_selection_geojson(
+        db: Session,
+        dataset: Dataset,
+        bbox: dict[str, Any],
+        *,
+        partition_scope_key: str,
+        area_id: uuid.UUID | None = None,
+        limit: int = 1000,
+        name: str | None = None,
+    ) -> ExportCreateResponse:
+        if dataset.source_name != "vmm_vha_bathymetry_profiles" or partition_scope_key != "flanders":
+            raise AppError(
+                code="PARTITIONED_VECTOR_EXPORT_UNSUPPORTED",
+                message="This vector source does not expose a governed partitioned export",
+                details={
+                    "source_name": dataset.source_name,
+                    "partition_scope_key": partition_scope_key,
+                },
+                status_code=400,
+            )
+
+        selection_geometry = None
+        partition_area_id = None
+        if area_id is not None:
+            area = db.get(Area, area_id)
+            if not area or area.project_id != dataset.project_id:
+                raise AppError(code="AREA_NOT_FOUND", message="Area not found", status_code=404)
+            selection_geometry, _covers_full_area = VectorFeatureService.constrain_bbox_to_area(
+                bbox,
+                area.geometry,
+            )
+            if str(area.name or "").lower().startswith("gemeente "):
+                partition_area_id = area.id
+
+        selection = VectorFeatureService.select_partitioned_features_by_bbox(
+            db,
+            project_id=dataset.project_id,
+            source_name=dataset.source_name,
+            partition_scope_key=partition_scope_key,
+            bbox=bbox,
+            limit=limit,
+            selection_geometry=selection_geometry,
+            selection_area_id=area_id,
+            partition_area_id=partition_area_id,
+        )
+        filename = ExportService._filename(name, "bathymetry-profile-selection.geojson", ".geojson")
+        export_path = StorageService.dataset_export_path(
+            str(dataset.project_id),
+            str(dataset.id),
+            filename,
+        )
+        metadata = {
+            "source": "partitioned_vector_selection",
+            "project_id": str(dataset.project_id),
+            "representative_dataset_id": str(dataset.id),
+            "dataset_ids": [str(value) for value in selection["dataset_ids"]],
+            "source_name": dataset.source_name,
+            "partition_scope_key": partition_scope_key,
+            "partition_count": selection["partition_count"],
+            "available_partition_count": selection["available_partition_count"],
+            "selection_bbox": selection["selection_bbox"],
+            "selection_area_id": selection.get("selection_area_id"),
+            "feature_count": selection["feature_count"],
+            "total_feature_count": selection["total_feature_count"],
+            "limit": selection["limit"],
+            "truncated": selection["truncated"],
+            "source_table": "vector_features",
+            "server_recomputed": True,
+        }
+        export = ExportService._write_json_export(
+            db,
+            project_id=dataset.project_id,
+            analysis_run_id=None,
+            export_type="partitioned_vector_selection_geojson",
+            storage_path=export_path,
+            content=selection["geojson"],
+            metadata=metadata,
+        )
+        return ExportService._create_response(export)
+
     @staticmethod
     def export_vector_selection_geojson(
         db: Session,

backend/app/services/vector_feature_service.py

@@ -137,6 +137,8 @@ SELECTION_FILTER_VALUE_ALIASES: dict[tuple[str, str], tuple[str, ...]] = {
 
 
 class VectorFeatureService:
+    MAX_VECTOR_PARTITIONS = 500
+
     @staticmethod
     def _expanded_selection_filter_values(filter_property: str, filter_values: list[Any]) -> list[str]:
         expanded: list[str] = []
@@ -307,6 +309,178 @@ class VectorFeatureService:
 
         return {"min_x": min_x, "min_y": min_y, "max_x": max_x, "max_y": max_y, "crs": "EPSG:4326"}
 
+    @staticmethod
+    def _dataset_bbox_intersects(
+        dataset: Dataset,
+        bbox: dict[str, float | str],
+    ) -> bool:
+        source_metadata = dataset.source_metadata if isinstance(dataset.source_metadata, dict) else {}
+        source_bbox = source_metadata.get("bbox_epsg4326")
+        if not isinstance(source_bbox, list) or len(source_bbox) != 4:
+            return True
+        try:
+            min_x, min_y, max_x, max_y = (float(value) for value in source_bbox)
+        except (TypeError, ValueError):
+            return True
+        return not (
+            max_x <= float(bbox["min_x"])
+            or min_x >= float(bbox["max_x"])
+            or max_y <= float(bbox["min_y"])
+            or min_y >= float(bbox["max_y"])
+        )
+
+    @staticmethod
+    def _latest_complete_partition_manifest(
+        datasets: Iterable[Dataset],
+        *,
+        source_name: str,
+        partition_scope_key: str,
+    ) -> list[Dataset]:
+        groups: dict[str, list[Dataset]] = {}
+        for dataset in datasets:
+            source_metadata = dataset.source_metadata if isinstance(dataset.source_metadata, dict) else {}
+            manifest_sha256 = str(source_metadata.get("partition_manifest_sha256") or "")
+            if (
+                dataset.source_name != source_name
+                or dataset.dataset_type not in {"vector", "geojson"}
+                or dataset.status != "ready"
+                or dataset.area_id is None
+                or source_metadata.get("regional_partitions_complete") is not True
+                or source_metadata.get("partition_scope_key") != partition_scope_key
+                or len(manifest_sha256) != 64
+            ):
+                continue
+            groups.setdefault(manifest_sha256, []).append(dataset)
+
+        complete_groups: list[list[Dataset]] = []
+        for group in groups.values():
+            area_ids = {dataset.area_id for dataset in group}
+            expected_data_count = max(
+                int((dataset.source_metadata or {}).get("data_partition_count") or 0)
+                for dataset in group
+            )
+            if expected_data_count > 0 and len(group) == expected_data_count and len(area_ids) == len(group):
+                complete_groups.append(group)
+
+        if not complete_groups:
+            return []
+
+        def manifest_priority(group: list[Dataset]) -> tuple[str, int, str]:
+            observed_at = max(
+                str((dataset.source_metadata or {}).get("partition_manifest_observed_at") or "")
+                for dataset in group
+            )
+            manifest_sha256 = str((group[0].source_metadata or {}).get("partition_manifest_sha256") or "")
+            return observed_at, len(group), manifest_sha256
+
+        selected = max(complete_groups, key=manifest_priority)
+        return sorted(selected, key=lambda dataset: (str(dataset.area_id), str(dataset.id)))
+
+    @staticmethod
+    def select_partitioned_features_by_bbox(
+        db,
+        *,
+        project_id: UUID,
+        source_name: str,
+        partition_scope_key: str,
+        bbox: dict[str, Any],
+        limit: int = 100,
+        selection_geometry: Any | None = None,
+        selection_area_id: UUID | None = None,
+        partition_area_id: UUID | None = None,
+    ) -> dict[str, Any]:
+        normalized_bbox = VectorFeatureService._normalize_selection_bbox(bbox)
+        safe_limit = max(1, min(int(limit), 1000))
+        project_datasets = (
+            db.query(Dataset)
+            .filter(
+                Dataset.project_id == project_id,
+                Dataset.source_name == source_name,
+                Dataset.status == "ready",
+            )
+            .all()
+        )
+        manifest_datasets = VectorFeatureService._latest_complete_partition_manifest(
+            project_datasets,
+            source_name=source_name,
+            partition_scope_key=partition_scope_key,
+        )
+        if not manifest_datasets:
+            raise AppError(
+                code="VECTOR_PARTITIONS_NOT_READY",
+                message="No complete persisted vector partition manifest is available",
+                details={"source_name": source_name, "partition_scope_key": partition_scope_key},
+                status_code=409,
+            )
+        if len(manifest_datasets) > VectorFeatureService.MAX_VECTOR_PARTITIONS:
+            raise AppError(
+                code="VECTOR_PARTITION_LIMIT_EXCEEDED",
+                message="The complete vector partition manifest exceeds the safety limit",
+                details={
+                    "partition_count": len(manifest_datasets),
+                    "max_partitions": VectorFeatureService.MAX_VECTOR_PARTITIONS,
+                },
+                status_code=422,
+            )
+
+        scoped_datasets = [
+            dataset
+            for dataset in manifest_datasets
+            if (partition_area_id is None or dataset.area_id == partition_area_id)
+            and VectorFeatureService._dataset_bbox_intersects(dataset, normalized_bbox)
+        ]
+        dataset_ids = [dataset.id for dataset in scoped_datasets]
+        representative = scoped_datasets[0] if scoped_datasets else manifest_datasets[0]
+        selection_shape = selection_geometry
+        if selection_shape is None:
+            selection_shape = ST_MakeEnvelope(
+                normalized_bbox["min_x"],
+                normalized_bbox["min_y"],
+                normalized_bbox["max_x"],
+                normalized_bbox["max_y"],
+                4326,
+            )
+
+        query = db.query(VectorFeature).filter(
+            VectorFeature.dataset_id.in_(dataset_ids),
+            ST_Intersects(VectorFeature.geometry, selection_shape),
+        )
+        total_feature_count = int(query.count())
+        rows = (
+            query.order_by(VectorFeature.created_at.asc(), VectorFeature.id.asc())
+            .limit(safe_limit + 1)
+            .all()
+        )
+        features = [
+            VectorFeatureService._row_to_geojson_feature(row)
+            for row in rows[:safe_limit]
+        ]
+        result = {
+            "selection_bbox": normalized_bbox,
+            "feature_count": len(features),
+            "total_feature_count": total_feature_count,
+            "limit": safe_limit,
+            "truncated": total_feature_count > safe_limit,
+            "geojson": {"type": "FeatureCollection", "features": features},
+            "partition_count": len(scoped_datasets),
+            "available_partition_count": len(manifest_datasets),
+            "partition_scope_key": partition_scope_key,
+            "source_name": source_name,
+            "dataset_ids": dataset_ids,
+        }
+        if selection_area_id is not None:
+            result["selection_area_id"] = str(selection_area_id)
+        if VectorFeatureService.supports_selection_summary(representative):
+            result["summary"] = VectorFeatureService.summarize_features_by_bbox(
+                db,
+                dataset=representative,
+                dataset_ids=dataset_ids,
+                bbox=normalized_bbox,
+                total_feature_count=total_feature_count,
+                selection_geometry=selection_shape,
+            )
+        return result
+
     @staticmethod
     def _row_to_geojson_feature(row: VectorFeature) -> dict[str, Any]:
         geometry_value = row.geometry
@@ -414,6 +588,7 @@ class VectorFeatureService:
         *,
         dataset: Dataset,
         bbox: dict[str, Any],
+        dataset_ids: list[UUID] | None = None,
         total_feature_count: int | None = None,
         selection_geometry: Any | None = None,
         full_dataset_area: bool = False,
@@ -429,7 +604,11 @@ class VectorFeatureService:
                 normalized_bbox["max_y"],
                 4326,
             )
-        selection_filter = (VectorFeature.dataset_id == dataset.id,)
+        selection_filter = (
+            (VectorFeature.dataset_id.in_(dataset_ids),)
+            if dataset_ids is not None
+            else (VectorFeature.dataset_id == dataset.id,)
+        )
         if preclipped_partition_filter is not None:
             partition_property, partition_value = preclipped_partition_filter
             selection_filter += (

backend/tests/test_sprint233_operational_completion.py

@@ -106,6 +106,56 @@ def test_current_vector_map_result_uses_authoritative_selection_export(monkeypat
     assert captured["limit"] == 1000
 
 
+def test_partitioned_vector_map_result_uses_governed_partition_export(monkeypatch) -> None:
+    project_id = uuid4()
+    dataset_id = uuid4()
+    dataset = Dataset(
+        id=dataset_id,
+        project_id=project_id,
+        name="vha-municipality.geojson",
+        dataset_type="vector",
+        source="VHA",
+        source_name="vmm_vha_bathymetry_profiles",
+        status="ready",
+    )
+    db = FakeSession({(Dataset, dataset_id): dataset})
+    expected = ExportCreateResponse(
+        export_id=uuid4(),
+        path="storage/exports/bathymetry-profile-selection.geojson",
+        status="ready",
+        export_type="partitioned_vector_selection_geojson",
+    )
+    captured: dict = {}
+
+    def fake_partition_export(*args, **kwargs):
+        captured["dataset"] = args[1]
+        captured.update(kwargs)
+        return expected
+
+    monkeypatch.setattr(
+        ExportService,
+        "export_partitioned_vector_selection_geojson",
+        fake_partition_export,
+    )
+    response = ExportService.export_map_result(
+        db,
+        MapResultExportRequest(
+            project_id=project_id,
+            mode="current",
+            dataset_id=dataset_id,
+            bbox=bbox_payload(),
+            partitioned=True,
+            partition_scope_key="flanders",
+            theme_id="bathymetry",
+        ),
+    )
+
+    assert response is expected
+    assert captured["dataset"] is dataset
+    assert captured["partition_scope_key"] == "flanders"
+    assert captured["limit"] == 1000
+
+
 def test_raster_map_result_is_recomputed_and_persisted(tmp_path, monkeypatch) -> None:
     project_id = uuid4()
     dataset_id = uuid4()

backend/tests/test_sprint236_bathymetry_expansion.py

@@ -10,7 +10,9 @@ from urllib.error import URLError
 from uuid import uuid4
 
 from fastapi.testclient import TestClient
+from geoalchemy2.shape import from_shape
 import pytest
+from shapely.geometry import MultiPolygon, Polygon
 
 from app.core.config import Settings
 from app.core.errors import AppError
@@ -20,6 +22,7 @@ from app.models import Area, Dataset, DatasetVersion, Project
 from app.schemas.bathymetry import BathymetryPartitionFinalizeRequest
 from app.services.bathymetry_profile_acquisition_service import BathymetryProfileAcquisitionService
 from app.services.mdk_bathymetry_probe_service import MdkBathymetryProbeService
+from app.services.vector_feature_service import VectorFeatureService
 
 
 ROOT = Path(__file__).resolve().parents[2]
@@ -247,6 +250,117 @@ def test_partition_finalization_requires_complete_area_accounting_and_updates_ve
     assert exc_info.value.code == "BATHYMETRY_PARTITION_MANIFEST_INCOMPLETE"
 
 
+def test_partition_selection_uses_latest_complete_manifest_without_duplicates() -> None:
+    project_id = uuid4()
+    source_name = BathymetryProfileAcquisitionService.PROVIDER
+
+    def partition(area_id, manifest, observed_at, data_partition_count):
+        return Dataset(
+            id=uuid4(),
+            project_id=project_id,
+            area_id=area_id,
+            name="vha.geojson",
+            dataset_type="vector",
+            source="VHA",
+            source_name=source_name,
+            source_metadata={
+                "regional_partitions_complete": True,
+                "partition_scope_key": "flanders",
+                "partition_manifest_sha256": manifest,
+                "partition_manifest_observed_at": observed_at,
+                "data_partition_count": data_partition_count,
+            },
+            status="ready",
+        )
+
+    old = [partition(uuid4(), "a" * 64, "2026-07-16T00:00:00+00:00", 1)]
+    new_area_ids = [uuid4(), uuid4()]
+    new = [
+        partition(area_id, "b" * 64, "2026-07-17T00:00:00+00:00", 2)
+        for area_id in new_area_ids
+    ]
+    incomplete = [partition(uuid4(), "c" * 64, "2026-07-18T00:00:00+00:00", 2)]
+
+    selected = VectorFeatureService._latest_complete_partition_manifest(
+        [*old, *new, *incomplete],
+        source_name=source_name,
+        partition_scope_key="flanders",
+    )
+
+    assert {dataset.area_id for dataset in selected} == set(new_area_ids)
+    assert all(dataset.source_metadata["partition_manifest_sha256"] == "b" * 64 for dataset in selected)
+
+
+def test_partition_selection_route_uses_exact_municipality_and_canonical_envelope(monkeypatch) -> None:
+    project_id, area_id, dataset_id = uuid4(), uuid4(), uuid4()
+    area_geometry = MultiPolygon(
+        [
+            Polygon(
+                [
+                    (5.0, 51.0),
+                    (5.2, 51.0),
+                    (5.2, 51.2),
+                    (5.0, 51.2),
+                    (5.0, 51.0),
+                ]
+            )
+        ]
+    )
+    area = Area(
+        id=area_id,
+        project_id=project_id,
+        name="Gemeente Mol - officiele grens",
+        geometry=from_shape(area_geometry, srid=4326),
+    )
+    db = SimpleNamespace(get=lambda model, row_id: area if model is Area and row_id == area_id else None)
+    captured = {}
+
+    def select_partitions(_db, **kwargs):
+        captured.update(kwargs)
+        return {
+            "selection_bbox": kwargs["bbox"],
+            "selection_area_id": area_id,
+            "feature_count": 1,
+            "total_feature_count": 1,
+            "limit": kwargs["limit"],
+            "truncated": False,
+            "geojson": {"type": "FeatureCollection", "features": []},
+            "partition_count": 1,
+            "available_partition_count": 269,
+            "partition_scope_key": "flanders",
+            "source_name": BathymetryProfileAcquisitionService.PROVIDER,
+            "dataset_ids": [dataset_id],
+        }
+
+    monkeypatch.setattr(VectorFeatureService, "select_partitioned_features_by_bbox", select_partitions)
+    app.dependency_overrides[get_db] = lambda: db
+    try:
+        response = TestClient(app).post(
+            f"/api/v1/projects/{project_id}/datasets/bathymetry/profiles/partitions/select",
+            json={
+                "bbox": {
+                    "min_x": 5.0,
+                    "min_y": 51.0,
+                    "max_x": 5.2,
+                    "max_y": 51.2,
+                    "crs": "EPSG:4326",
+                },
+                "area_id": str(area_id),
+                "limit": 1000,
+            },
+        )
+    finally:
+        app.dependency_overrides.clear()
+
+    assert response.status_code == 200
+    assert set(response.json()) == {"data"}
+    assert response.json()["data"]["partition_count"] == 1
+    assert response.json()["data"]["available_partition_count"] == 269
+    assert captured["partition_area_id"] == area_id
+    assert captured["source_name"] == BathymetryProfileAcquisitionService.PROVIDER
+    assert captured["partition_scope_key"] == "flanders"
+
+
 def test_flanders_scope_discovery_uses_complete_unique_vrbg_inventory() -> None:
     features = [
         {
@@ -333,3 +447,22 @@ def test_frontend_labels_flanders_scope_without_kempen_mislabeling() -> None:
     assert "'Grens Vlaanderen'" in dataset_display
     assert "coverageScope === 'flanders' && layer === 'municipality_boundaries'" in dataset_display
     assert "'Gemeentegrenzen Vlaanderen'" in dataset_display
+
+
+def test_frontend_uses_partitioned_bathymetry_selection_for_regional_scope() -> None:
+    map_workspace = (
+        ROOT / "frontend" / "src" / "components" / "map" / "MapWorkspace.tsx"
+    ).read_text(encoding="utf-8")
+    theme_hook = (
+        ROOT / "frontend" / "src" / "hooks" / "useMapThemeSelectionInsights.ts"
+    ).read_text(encoding="utf-8")
+    dataset_api = (
+        ROOT / "frontend" / "src" / "services" / "api" / "datasets.ts"
+    ).read_text(encoding="utf-8")
+
+    assert "isPartitionedBathymetry" in map_workspace
+    assert "regionalPartitionedThemeActive" in map_workspace
+    assert "datasetAvailabilityLabel(dataset, partitions)" in map_workspace
+    assert "activeSelectionResult?.geojson ?? null" in map_workspace
+    assert "selectBathymetryProfilePartitions" in theme_hook
+    assert "datasets/bathymetry/profiles/partitions/select" in dataset_api