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