nominatim/db/sqlalchemy_functions.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2023 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Custom functions and expressions for SQLAlchemy.
   9 """
  10 from __future__ import annotations
  11 from typing import Any
  12
  13 import sqlalchemy as sa
  14 from sqlalchemy.ext.compiler import compiles
  15
  16 from nominatim.typing import SaColumn
  17
  18 # pylint: disable=all
  19
  20 class PlacexGeometryReverseLookuppolygon(sa.sql.functions.GenericFunction[Any]):
  21     """ Check for conditions that allow partial index use on
  22         'idx_placex_geometry_reverse_lookupPolygon'.
  23
  24         Needs to be constant, so that the query planner picks them up correctly
  25         in prepared statements.
  26     """
  27     name = 'PlacexGeometryReverseLookuppolygon'
  28     inherit_cache = True
  29
  30
  31 @compiles(PlacexGeometryReverseLookuppolygon) # type: ignore[no-untyped-call, misc]
  32 def _default_intersects(element: SaColumn,
  33                         compiler: 'sa.Compiled', **kw: Any) -> str:
  34     return ("(ST_GeometryType(placex.geometry) in ('ST_Polygon', 'ST_MultiPolygon')"
  35             " AND placex.rank_address between 4 and 25"
  36             " AND placex.type != 'postcode'"
  37             " AND placex.name is not null"
  38             " AND placex.indexed_status = 0"
  39             " AND placex.linked_place_id is null)")
  40
  41
  42 @compiles(PlacexGeometryReverseLookuppolygon, 'sqlite') # type: ignore[no-untyped-call, misc]
  43 def _sqlite_intersects(element: SaColumn,
  44                        compiler: 'sa.Compiled', **kw: Any) -> str:
  45     return ("(ST_GeometryType(placex.geometry) in ('POLYGON', 'MULTIPOLYGON')"
  46             " AND placex.rank_address between 4 and 25"
  47             " AND placex.type != 'postcode'"
  48             " AND placex.name is not null"
  49             " AND placex.indexed_status = 0"
  50             " AND placex.linked_place_id is null)")
  51
  52
  53 class IntersectsReverseDistance(sa.sql.functions.GenericFunction[Any]):
  54     name = 'IntersectsReverseDistance'
  55     inherit_cache = True
  56
  57     def __init__(self, table: sa.Table, geom: SaColumn) -> None:
  58         super().__init__(table.c.geometry, # type: ignore[no-untyped-call]
  59                          table.c.rank_search, geom)
  60         self.tablename = table.name
  61
  62
  63 @compiles(IntersectsReverseDistance) # type: ignore[no-untyped-call, misc]
  64 def default_reverse_place_diameter(element: SaColumn,
  65                                    compiler: 'sa.Compiled', **kw: Any) -> str:
  66     table = element.tablename
  67     return f"({table}.rank_address between 4 and 25"\
  68            f" AND {table}.type != 'postcode'"\
  69            f" AND {table}.name is not null"\
  70            f" AND {table}.linked_place_id is null"\
  71            f" AND {table}.osm_type = 'N'" + \
  72            " AND ST_Buffer(%s, reverse_place_diameter(%s)) && %s)" % \
  73                tuple(map(lambda c: compiler.process(c, **kw), element.clauses))
  74
  75
  76 @compiles(IntersectsReverseDistance, 'sqlite') # type: ignore[no-untyped-call, misc]
  77 def sqlite_reverse_place_diameter(element: SaColumn,
  78                                   compiler: 'sa.Compiled', **kw: Any) -> str:
  79     geom1, rank, geom2 = list(element.clauses)
  80     table = element.tablename
  81
  82     return (f"({table}.rank_address between 4 and 25"\
  83             f" AND {table}.type != 'postcode'"\
  84             f" AND {table}.name is not null"\
  85             f" AND {table}.linked_place_id is null"\
  86             f" AND {table}.osm_type = 'N'"\
  87              " AND MbrIntersects(%s, ST_Expand(%s, 14.0 * exp(-0.2 * %s) - 0.03))"\
  88             f" AND {table}.place_id IN"\
  89              " (SELECT place_id FROM placex_place_node_areas"\
  90              "  WHERE ROWID IN (SELECT ROWID FROM SpatialIndex"\
  91              "  WHERE f_table_name = 'placex_place_node_areas'"\
  92              "  AND search_frame = %s)))") % (
  93                 compiler.process(geom1, **kw),
  94                 compiler.process(geom2, **kw),
  95                 compiler.process(rank, **kw),
  96                 compiler.process(geom2, **kw))
  97
  98
  99 class IsBelowReverseDistance(sa.sql.functions.GenericFunction[Any]):
 100     name = 'IsBelowReverseDistance'
 101     inherit_cache = True
 102
 103
 104 @compiles(IsBelowReverseDistance) # type: ignore[no-untyped-call, misc]
 105 def default_is_below_reverse_distance(element: SaColumn,
 106                                       compiler: 'sa.Compiled', **kw: Any) -> str:
 107     dist, rank = list(element.clauses)
 108     return "%s < reverse_place_diameter(%s)" % (compiler.process(dist, **kw),
 109                                                 compiler.process(rank, **kw))
 110
 111
 112 @compiles(IsBelowReverseDistance, 'sqlite') # type: ignore[no-untyped-call, misc]
 113 def sqlite_is_below_reverse_distance(element: SaColumn,
 114                                      compiler: 'sa.Compiled', **kw: Any) -> str:
 115     dist, rank = list(element.clauses)
 116     return "%s < 14.0 * exp(-0.2 * %s) - 0.03" % (compiler.process(dist, **kw),
 117                                                   compiler.process(rank, **kw))
 118
 119
 120 def select_index_placex_geometry_reverse_lookupplacenode(table: str) -> 'sa.TextClause':
 121     """ Create an expression with the necessary conditions over a placex
 122         table that the index 'idx_placex_geometry_reverse_lookupPlaceNode'
 123         can be used.
 124     """
 125     return sa.text(f"{table}.rank_address between 4 and 25"
 126                    f" AND {table}.type != 'postcode'"
 127                    f" AND {table}.name is not null"
 128                    f" AND {table}.linked_place_id is null"
 129                    f" AND {table}.osm_type = 'N'")
 130
 131
 132 class IsAddressPoint(sa.sql.functions.GenericFunction[Any]):
 133     name = 'IsAddressPoint'
 134     inherit_cache = True
 135
 136     def __init__(self, table: sa.Table) -> None:
 137         super().__init__(table.c.rank_address, # type: ignore[no-untyped-call]
 138                          table.c.housenumber, table.c.name)
 139
 140
 141 @compiles(IsAddressPoint) # type: ignore[no-untyped-call, misc]
 142 def default_is_address_point(element: SaColumn,
 143                              compiler: 'sa.Compiled', **kw: Any) -> str:
 144     rank, hnr, name = list(element.clauses)
 145     return "(%s = 30 AND (%s IS NOT NULL OR %s ? 'addr:housename'))" % (
 146                 compiler.process(rank, **kw),
 147                 compiler.process(hnr, **kw),
 148                 compiler.process(name, **kw))
 149
 150
 151 @compiles(IsAddressPoint, 'sqlite') # type: ignore[no-untyped-call, misc]
 152 def sqlite_is_address_point(element: SaColumn,
 153                             compiler: 'sa.Compiled', **kw: Any) -> str:
 154     rank, hnr, name = list(element.clauses)
 155     return "(%s = 30 AND coalesce(%s, json_extract(%s, '$.addr:housename')) IS NOT NULL)" % (
 156                 compiler.process(rank, **kw),
 157                 compiler.process(hnr, **kw),
 158                 compiler.process(name, **kw))
 159
 160
 161 class CrosscheckNames(sa.sql.functions.GenericFunction[Any]):
 162     """ Check if in the given list of names in parameters 1 any of the names
 163         from the JSON array in parameter 2 are contained.
 164     """
 165     name = 'CrosscheckNames'
 166     inherit_cache = True
 167
 168 @compiles(CrosscheckNames) # type: ignore[no-untyped-call, misc]
 169 def compile_crosscheck_names(element: SaColumn,
 170                              compiler: 'sa.Compiled', **kw: Any) -> str:
 171     arg1, arg2 = list(element.clauses)
 172     return "coalesce(avals(%s) && ARRAY(SELECT * FROM json_array_elements_text(%s)), false)" % (
 173             compiler.process(arg1, **kw), compiler.process(arg2, **kw))
 174
 175
 176 @compiles(CrosscheckNames, 'sqlite') # type: ignore[no-untyped-call, misc]
 177 def compile_sqlite_crosscheck_names(element: SaColumn,
 178                                     compiler: 'sa.Compiled', **kw: Any) -> str:
 179     arg1, arg2 = list(element.clauses)
 180     return "EXISTS(SELECT *"\
 181            " FROM json_each(%s) as name, json_each(%s) as match_name"\
 182            " WHERE name.value = match_name.value)"\
 183            % (compiler.process(arg1, **kw), compiler.process(arg2, **kw))
 184
 185
 186 class JsonArrayEach(sa.sql.functions.GenericFunction[Any]):
 187     """ Return elements of a json array as a set.
 188     """
 189     name = 'JsonArrayEach'
 190     inherit_cache = True
 191
 192
 193 @compiles(JsonArrayEach) # type: ignore[no-untyped-call, misc]
 194 def default_json_array_each(element: SaColumn, compiler: 'sa.Compiled', **kw: Any) -> str:
 195     return "json_array_elements(%s)" % compiler.process(element.clauses, **kw)
 196
 197
 198 @compiles(JsonArrayEach, 'sqlite') # type: ignore[no-untyped-call, misc]
 199 def sqlite_json_array_each(element: SaColumn, compiler: 'sa.Compiled', **kw: Any) -> str:
 200     return "json_each(%s)" % compiler.process(element.clauses, **kw)
 201
 202
 203 class Greatest(sa.sql.functions.GenericFunction[Any]):
 204     """ Function to compute maximum of all its input parameters.
 205     """
 206     name = 'greatest'
 207     inherit_cache = True
 208
 209
 210 @compiles(Greatest, 'sqlite') # type: ignore[no-untyped-call, misc]
 211 def sqlite_greatest(element: SaColumn, compiler: 'sa.Compiled', **kw: Any) -> str:
 212     return "max(%s)" % compiler.process(element.clauses, **kw)