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) 2025 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, SaFromClause
  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.name is not null"
  36             " AND placex.indexed_status = 0"
  37             " AND placex.linked_place_id is null)")
  38
  39
  40 @compiles(PlacexGeometryReverseLookuppolygon, 'sqlite')
  41 def _sqlite_intersects(element: PlacexGeometryReverseLookuppolygon,
  42                        compiler: 'sa.Compiled', **kw: Any) -> str:
  43     return ("(ST_GeometryType(placex.geometry) in ('POLYGON', 'MULTIPOLYGON')"
  44             " AND placex.rank_address between 4 and 25"
  45             " AND placex.name is not null"
  46             " AND placex.indexed_status = 0"
  47             " AND placex.linked_place_id is null)")
  48
  49
  50 class IntersectsReverseDistance(sa.sql.functions.GenericFunction[Any]):
  51     name = 'IntersectsReverseDistance'
  52     inherit_cache = True
  53
  54     def __init__(self, table: sa.Table, geom: SaColumn) -> None:
  55         super().__init__(table.c.geometry,
  56                          table.c.rank_search, geom)
  57         self.tablename = table.name
  58
  59
  60 @compiles(IntersectsReverseDistance)
  61 def default_reverse_place_diameter(element: IntersectsReverseDistance,
  62                                    compiler: 'sa.Compiled', **kw: Any) -> str:
  63     table = element.tablename
  64     return f"({table}.rank_address between 4 and 25"\
  65            f" AND {table}.name is not null"\
  66            f" AND {table}.linked_place_id is null"\
  67            f" AND {table}.osm_type = 'N'" + \
  68            " AND ST_Buffer(%s, reverse_place_diameter(%s)) && %s)" \
  69         % tuple(map(lambda c: compiler.process(c, **kw), element.clauses))
  70
  71
  72 @compiles(IntersectsReverseDistance, 'sqlite')
  73 def sqlite_reverse_place_diameter(element: IntersectsReverseDistance,
  74                                   compiler: 'sa.Compiled', **kw: Any) -> str:
  75     geom1, rank, geom2 = list(element.clauses)
  76     table = element.tablename
  77
  78     return (f"({table}.rank_address between 4 and 25"
  79             f" AND {table}.name is not null"
  80             f" AND {table}.linked_place_id is null"
  81             f" AND {table}.osm_type = 'N'"
  82             "  AND MbrIntersects(%s, ST_Expand(%s, 14.0 * exp(-0.2 * %s) - 0.03))"
  83             f" AND {table}.place_id IN"
  84             "  (SELECT place_id FROM placex_place_node_areas"
  85             "   WHERE ROWID IN (SELECT ROWID FROM SpatialIndex"
  86             "   WHERE f_table_name = 'placex_place_node_areas'"
  87             "   AND search_frame = %s)))") % (
  88                 compiler.process(geom1, **kw),
  89                 compiler.process(geom2, **kw),
  90                 compiler.process(rank, **kw),
  91                 compiler.process(geom2, **kw))
  92
  93
  94 class IsBelowReverseDistance(sa.sql.functions.GenericFunction[Any]):
  95     name = 'IsBelowReverseDistance'
  96     inherit_cache = True
  97
  98
  99 @compiles(IsBelowReverseDistance)
 100 def default_is_below_reverse_distance(element: IsBelowReverseDistance,
 101                                       compiler: 'sa.Compiled', **kw: Any) -> str:
 102     dist, rank = list(element.clauses)
 103     return "%s < reverse_place_diameter(%s)" % (compiler.process(dist, **kw),
 104                                                 compiler.process(rank, **kw))
 105
 106
 107 @compiles(IsBelowReverseDistance, 'sqlite')
 108 def sqlite_is_below_reverse_distance(element: IsBelowReverseDistance,
 109                                      compiler: 'sa.Compiled', **kw: Any) -> str:
 110     dist, rank = list(element.clauses)
 111     return "%s < 14.0 * exp(-0.2 * %s) - 0.03" % (compiler.process(dist, **kw),
 112                                                   compiler.process(rank, **kw))
 113
 114
 115 class IsAddressPoint(sa.sql.functions.GenericFunction[Any]):
 116     name = 'IsAddressPoint'
 117     inherit_cache = True
 118
 119     def __init__(self, table: sa.Table) -> None:
 120         super().__init__(table.c.rank_address,
 121                          table.c.housenumber, table.c.name, table.c.address)
 122
 123
 124 @compiles(IsAddressPoint)
 125 def default_is_address_point(element: IsAddressPoint,
 126                              compiler: 'sa.Compiled', **kw: Any) -> str:
 127     rank, hnr, name, address = list(element.clauses)
 128     return "(%s = 30 AND (%s IS NULL OR NOT %s ? '_inherited')" \
 129            " AND (%s IS NOT NULL OR %s ? 'addr:housename'))" % (
 130                 compiler.process(rank, **kw),
 131                 compiler.process(address, **kw),
 132                 compiler.process(address, **kw),
 133                 compiler.process(hnr, **kw),
 134                 compiler.process(name, **kw))
 135
 136
 137 @compiles(IsAddressPoint, 'sqlite')
 138 def sqlite_is_address_point(element: IsAddressPoint,
 139                             compiler: 'sa.Compiled', **kw: Any) -> str:
 140     rank, hnr, name, address = list(element.clauses)
 141     return "(%s = 30 AND json_extract(%s, '$._inherited') IS NULL" \
 142            " AND coalesce(%s, json_extract(%s, '$.addr:housename')) IS NOT NULL)" % (
 143                 compiler.process(rank, **kw),
 144                 compiler.process(address, **kw),
 145                 compiler.process(hnr, **kw),
 146                 compiler.process(name, **kw))
 147
 148
 149 class CrosscheckNames(sa.sql.functions.GenericFunction[Any]):
 150     """ Check if in the given list of names in parameters 1 any of the names
 151         from the JSON array in parameter 2 are contained.
 152     """
 153     name = 'CrosscheckNames'
 154     inherit_cache = True
 155
 156
 157 @compiles(CrosscheckNames)
 158 def compile_crosscheck_names(element: CrosscheckNames,
 159                              compiler: 'sa.Compiled', **kw: Any) -> str:
 160     arg1, arg2 = list(element.clauses)
 161     return "coalesce(avals(%s) && ARRAY(SELECT * FROM json_array_elements_text(%s)), false)" % (
 162             compiler.process(arg1, **kw), compiler.process(arg2, **kw))
 163
 164
 165 @compiles(CrosscheckNames, 'sqlite')
 166 def compile_sqlite_crosscheck_names(element: CrosscheckNames,
 167                                     compiler: 'sa.Compiled', **kw: Any) -> str:
 168     arg1, arg2 = list(element.clauses)
 169     return "EXISTS(SELECT *"\
 170            " FROM json_each(%s) as name, json_each(%s) as match_name"\
 171            " WHERE name.value = match_name.value)"\
 172            % (compiler.process(arg1, **kw), compiler.process(arg2, **kw))
 173
 174
 175 class JsonArrayEach(sa.sql.functions.GenericFunction[Any]):
 176     """ Return elements of a json array as a set.
 177     """
 178     name = 'JsonArrayEach'
 179     inherit_cache = True
 180
 181
 182 @compiles(JsonArrayEach)
 183 def default_json_array_each(element: JsonArrayEach, compiler: 'sa.Compiled', **kw: Any) -> str:
 184     return "json_array_elements(%s)" % compiler.process(element.clauses, **kw)
 185
 186
 187 @compiles(JsonArrayEach, 'sqlite')
 188 def sqlite_json_array_each(element: JsonArrayEach, compiler: 'sa.Compiled', **kw: Any) -> str:
 189     return "json_each(%s)" % compiler.process(element.clauses, **kw)
 190
 191
 192 class Greatest(sa.sql.functions.GenericFunction[Any]):
 193     """ Function to compute maximum of all its input parameters.
 194     """
 195     name = 'greatest'
 196     inherit_cache = True
 197
 198
 199 @compiles(Greatest, 'sqlite')
 200 def sqlite_greatest(element: Greatest, compiler: 'sa.Compiled', **kw: Any) -> str:
 201     return "max(%s)" % compiler.process(element.clauses, **kw)
 202
 203
 204 class RegexpWord(sa.sql.functions.GenericFunction[Any]):
 205     """ Check if a full word is in a given string.
 206     """
 207     name = 'RegexpWord'
 208     inherit_cache = True
 209
 210
 211 @compiles(RegexpWord, 'postgresql')
 212 def postgres_regexp_nocase(element: RegexpWord, compiler: 'sa.Compiled', **kw: Any) -> str:
 213     arg1, arg2 = list(element.clauses)
 214     return "%s ~* ('\\m(' || %s  || ')\\M')::text" \
 215         % (compiler.process(arg2, **kw), compiler.process(arg1, **kw))
 216
 217
 218 @compiles(RegexpWord, 'sqlite')
 219 def sqlite_regexp_nocase(element: RegexpWord, compiler: 'sa.Compiled', **kw: Any) -> str:
 220     arg1, arg2 = list(element.clauses)
 221     return "regexp('\\b(' || %s  || ')\\b', %s)"\
 222         % (compiler.process(arg1, **kw), compiler.process(arg2, **kw))
 223
 224
 225 class CategoryMatch(sa.sql.functions.GenericFunction[Any]):
 226     """ Match a placex row against a category or any of its descendants.
 227
 228         On PostgreSQL this queries the ltree 'categories' column using the
 229         GiST index (``categories <@ '<category>'``). On SQLite, where the
 230         categories are stored as a comma-separated text, the same is
 231         emulated with a substring search.
 232     """
 233     name = 'CategoryMatch'
 234     inherit_cache = True
 235
 236     def __init__(self, table: SaFromClause, category: str) -> None:
 237         # The needles for the SQLite variant are precomputed here because
 238         # SQLAlchemy 1.4 binds a parameter only once per statement, even
 239         # when it is rendered more than once.
 240         super().__init__(table.c.categories, sa.literal(category),
 241                          sa.literal(f',{category},'), sa.literal(f',{category}.'))
 242
 243
 244 @compiles(CategoryMatch)
 245 def _default_category_match(element: CategoryMatch,
 246                             compiler: 'sa.Compiled', **kw: Any) -> str:
 247     cats, category, _, _ = list(element.clauses)
 248     return "(%s <@ (%s)::ltree)" % (compiler.process(cats, **kw),
 249                                     compiler.process(category, **kw))
 250
 251
 252 @compiles(CategoryMatch, 'sqlite')
 253 def _sqlite_category_match(element: CategoryMatch,
 254                            compiler: 'sa.Compiled', **kw: Any) -> str:
 255     cats, _, exact, descendants = list(element.clauses)
 256     # The categories are padded with the separator on both sides, so that
 257     # the needles match only on a full label boundary: ',<category>,' is an
 258     # exact hit, ',<category>.' one of its descendants.
 259     haystack = "(',' || %s || ',')" % compiler.process(cats, **kw)
 260     return "(instr(%s, %s) > 0 OR instr(%s, %s) > 0)" \
 261            % (haystack, compiler.process(exact, **kw),
 262               haystack, compiler.process(descendants, **kw))