Convert the Regions materialised-view to a table

jeremyh · jeremyh · commit 1a5f7b37f46f · 2020-09-10T14:25:06.000+10:00
It gives us a lot more flexibility. Such as handling
errors and incremental updating.

Users who were running previous dev versions of Explorer will
need to `--init` again.
diff --git a/cubedash/summary/_schema.py b/cubedash/summary/_schema.py
@@ -1,7 +1,9 @@
 from __future__ import absolute_import
 
-from typing import Set
+import warnings
 from enum import Enum
+from typing import Set
+
 import structlog
 from geoalchemy2 import Geometry
 from sqlalchemy import (
@@ -156,6 +158,24 @@
     ),
 )
 
+# The geometry of each unique 'region' for a product.
+REGION = Table(
+    "region",
+    METADATA,
+    Column("dataset_type_ref", SmallInteger, nullable=False),
+    Column("region_code", String, nullable=False),
+    Column("count", Integer, nullable=False),
+    Column(
+        "generation_time",
+        DateTime(timezone=True),
+        server_default=func.now(),
+        nullable=False,
+    ),
+    Column("footprint", Geometry(srid=4326, spatial_index=False)),
+    PrimaryKeyConstraint("dataset_type_ref", "region_code"),
+)
+
+
 _REF_TABLE_METADATA = MetaData(schema=CUBEDASH_SCHEMA)
 # This is a materialised view of the postgis spatial_ref_sys for lookups.
 # See creation of mv_spatial_ref_sys below.
@@ -182,16 +202,6 @@
     Column("has_region", Integer),
 )
 
-# The geometry of each unique 'region' for a product.
-REGION = Table(
-    "mv_region",
-    _REF_TABLE_METADATA,
-    Column("dataset_type_ref", SmallInteger, nullable=False),
-    Column("region_code", String),
-    Column("footprint", Geometry(srid=4326, spatial_index=False)),
-    Column("count", Integer, nullable=False),
-)
-
 
 def has_schema(engine: Engine) -> bool:
     """
@@ -207,6 +217,16 @@ def is_compatible_schema(engine: Engine) -> bool:
     if not pg_column_exists(engine, f"{CUBEDASH_SCHEMA}.product", "fixed_metadata"):
         is_latest = False
 
+    if not pg_exists(engine, f"{CUBEDASH_SCHEMA}.region"):
+        is_latest = False
+
+    if pg_exists(engine, f"{CUBEDASH_SCHEMA}.mv_region"):
+        warnings.warn(
+            "Your database has item `cubedash.mv_region` from an unstable version of Explorer. "
+            "It will not harm you, but feel free to drop it once all Explorer instances "
+            "have been upgraded: "
+            "    drop materialised view cubedash.mv_region"
+        )
     return is_latest
 
 
@@ -346,30 +366,6 @@ def create_schema(engine: Engine):
     """
     )
 
-    # A geometry for each declared region.
-    #
-    # This happens in two steps so that we union cleanly (in the native CRS rather than after transforming)
-    # TODO: Simplify geom after union?
-    engine.execute(
-        f"""
-    create materialized view if not exists {CUBEDASH_SCHEMA}.mv_region as (
-        select dataset_type_ref,
-               region_code,
-               ST_SimplifyPreserveTopology(ST_Union(footprint), 0.0001) as footprint,
-               sum(count)          as count
-        from (
-             select dataset_type_ref,
-                    region_code,
-                    ST_Transform(ST_Union(footprint), 4326) as footprint,
-                    count(*)                                as count
-             from {CUBEDASH_SCHEMA}.dataset_spatial
-             group by dataset_type_ref, region_code, ST_SRID(footprint)
-        ) srid_groups
-        group by dataset_type_ref, region_code
-    ) with no data;
-    """
-    )
-
 
 def refresh_supporting_views(conn, concurrently=False):
     args = "concurrently" if concurrently else ""
@@ -383,11 +379,6 @@ def refresh_supporting_views(conn, concurrently=False):
     refresh materialized view {args} {CUBEDASH_SCHEMA}.mv_dataset_spatial_quality;
     """
     )
-    conn.execute(
-        f"""
-    refresh materialized view {args} {CUBEDASH_SCHEMA}.mv_region;
-    """
-    )
 
 
 def get_srid_name(engine: Engine, srid: int):
diff --git a/cubedash/summary/_stores.py b/cubedash/summary/_stores.py
@@ -3,7 +3,17 @@
 from collections import Counter
 from dataclasses import dataclass
 from datetime import date, datetime, timedelta
-from typing import Dict, Generator, Iterable, List, Optional, Sequence, Tuple, Union
+from typing import (
+    Dict,
+    Generator,
+    Iterable,
+    List,
+    Optional,
+    Sequence,
+    Tuple,
+    Union,
+    Set,
+)
 from uuid import UUID
 
 import dateutil.parser
@@ -14,7 +24,7 @@
 from geoalchemy2 import shape as geo_shape
 from geoalchemy2.shape import to_shape
 from shapely.geometry import GeometryCollection
-from sqlalchemy import DDL, String, and_, func, select
+from sqlalchemy import DDL, String, and_, func, select, bindparam, SmallInteger, literal
 from sqlalchemy.dialects import postgresql as postgres
 from sqlalchemy.dialects.postgresql import TSTZRANGE
 from sqlalchemy.engine import Engine, RowProxy
@@ -128,13 +138,14 @@ def init(self):
 
         (Requires `create` permissions in the db)
         """
+        needed_update = not _schema.is_compatible_schema(self._engine)
+
+        # Add any missing schema items or patches.
         _schema.create_schema(self._engine)
-        # Apply any needed updates.
-        refresh_items = _schema.update_schema(self._engine)
+        refresh_also = _schema.update_schema(self._engine)
 
-        # Refresh relevant data summaries
-        for refresh_item in refresh_items:
-            _refresh_data(refresh_item, store=self)
+        if needed_update or refresh_also:
+            _refresh_data(refresh_also, store=self)
 
     @classmethod
     def create(cls, index: Index, log=_LOG) -> "SummaryStore":
@@ -185,7 +196,7 @@ def refresh_product(
             return None
 
         _LOG.info("init.product", product_name=product.name)
-        added_count = _extents.refresh_product(
+        change_count = _extents.refresh_product(
             self.index,
             product,
             recompute_all_extents=force_dataset_extent_recompute,
@@ -226,7 +237,62 @@ def refresh_product(
                 fixed_metadata=fixed_metadata,
             )
         )
-        return added_count
+
+        self._refresh_product_regions(product)
+        _LOG.info("init.regions.done", product_name=product.name)
+        return change_count
+
+    def _refresh_product_regions(self, dataset_type: DatasetType) -> int:
+        log = _LOG.bind(product_name=dataset_type.name)
+        log.info("refresh.regions.start")
+        select_by_srid = (
+            select(
+                [
+                    DATASET_SPATIAL.c.dataset_type_ref,
+                    DATASET_SPATIAL.c.region_code,
+                    func.ST_Transform(
+                        func.ST_Union(DATASET_SPATIAL.c.footprint), 4326
+                    ).label("footprint"),
+                    func.count().label("count"),
+                ]
+            )
+            .where(
+                DATASET_SPATIAL.c.dataset_type_ref
+                == bindparam("product_ref", dataset_type.id, type_=SmallInteger)
+            )
+            .group_by("dataset_type_ref", "region_code")
+            .cte("srid_groups")
+        )
+
+        columns = dict(
+            dataset_type_ref=select_by_srid.c.dataset_type_ref,
+            region_code=func.coalesce(select_by_srid.c.region_code, ""),
+            footprint=func.ST_SimplifyPreserveTopology(
+                func.ST_Union(select_by_srid.c.footprint), literal(0.0001)
+            ),
+            count=func.sum(select_by_srid.c.count),
+        )
+        query = postgres.insert(REGION).from_select(
+            columns.keys(),
+            select(columns.values())
+            .select_from(select_by_srid)
+            .group_by("dataset_type_ref", "region_code"),
+        )
+        query = query.on_conflict_do_update(
+            index_elements=["dataset_type_ref", "region_code"],
+            set_=dict(
+                footprint=query.excluded.footprint,
+                count=query.excluded.count,
+                generation_time=func.now(),
+            ),
+        )
+        # Path(__file__).parent.joinpath("insertion.sql").write_text(
+        #     f"\n{as_sql(query)}\n"
+        # )
+        changed_rows = self._engine.execute(query).rowcount
+
+        log.info("refresh.regions.end", changed_regions=changed_rows)
+        return changed_rows
 
     def refresh_stats(self, concurrently=False):
         """
@@ -905,21 +971,24 @@ def get_dataset_footprint_region(self, dataset_id):
         )
 
 
-def _refresh_data(item: PleaseRefresh, store: SummaryStore):
+def _refresh_data(please_refresh: Set[PleaseRefresh], store: SummaryStore):
     """
-    Refresh the given kind of data.
+    Refresh product information after a schema update, plus the given kind of data.
     """
-    if item == PleaseRefresh.DATASET_EXTENTS:
-        for dt in store.all_dataset_types():
-            _LOG.info("data.refreshing_extents", product=dt.name)
-            # Skip product if it's never been summarised at all.
-            if store.get_product_summary(dt.name) is None:
-                continue
-
-            store.refresh_product(dt, force_dataset_extent_recompute=True)
-        _LOG.info("data.refreshing_extents.complete")
-    else:
-        raise NotImplementedError(f"Unknown data type to refresh_data: {item}")
+    recompute_dataset_extents = PleaseRefresh.DATASET_EXTENTS in please_refresh
+
+    for dt in store.all_dataset_types():
+        _LOG.info("data.refreshing_extents", product=dt.name)
+        # Skip product if it's never been summarised at all.
+        if store.get_product_summary(dt.name) is None:
+            continue
+
+        store.refresh_product(
+            dt,
+            refresh_older_than=timedelta(minutes=-1),
+            force_dataset_extent_recompute=recompute_dataset_extents,
+        )
+    _LOG.info("data.refreshing_extents.complete")
 
 
 def _safe_read_date(d):
