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[bool]):
  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     type = sa.Boolean()
  28     name = 'PlacexGeometryReverseLookuppolygon'
  29     inherit_cache = True
  30
  31
  32 @compiles(PlacexGeometryReverseLookuppolygon) # type: ignore[no-untyped-call, misc]
  33 def _default_intersects(element: SaColumn,
  34                         compiler: 'sa.Compiled', **kw: Any) -> str:
  35     return ("(ST_GeometryType(placex.geometry) in ('ST_Polygon', 'ST_MultiPolygon')"
  36             " AND placex.rank_address between 4 and 25"
  37             " AND placex.type != 'postcode'"
  38             " AND placex.name is not null"
  39             " AND placex.indexed_status = 0"
  40             " AND placex.linked_place_id is null)")
  41
  42
  43 @compiles(PlacexGeometryReverseLookuppolygon, 'sqlite') # type: ignore[no-untyped-call, misc]
  44 def _sqlite_intersects(element: SaColumn,
  45                        compiler: 'sa.Compiled', **kw: Any) -> str:
  46     return ("(ST_GeometryType(placex.geometry) in ('POLYGON', 'MULTIPOLYGON')"
  47             " AND placex.rank_address between 4 and 25"
  48             " AND placex.type != 'postcode'"
  49             " AND placex.name is not null"
  50             " AND placex.indexed_status = 0"
  51             " AND placex.linked_place_id is null)")
  52
  53
  54 class IntersectsReverseDistance(sa.sql.functions.GenericFunction[bool]):
  55     type = sa.Boolean()
  56     name = 'IntersectsReverseDistance'
  57     inherit_cache = True
  58
  59     def __init__(self, table: sa.Table, geom: SaColumn) -> None:
  60         super().__init__(table.c.geometry, # type: ignore[no-untyped-call]
  61                          table.c.rank_search, geom)
  62         self.tablename = table.name
  63
  64
  65 @compiles(IntersectsReverseDistance) # type: ignore[no-untyped-call, misc]
  66 def default_reverse_place_diameter(element: SaColumn,
  67                                    compiler: 'sa.Compiled', **kw: Any) -> str:
  68     table = element.tablename
  69     return f"{table}.rank_address between 4 and 25"\
  70            f" AND {table}.type != 'postcode'"\
  71            f" AND {table}.name is not null"\
  72            f" AND {table}.linked_place_id is null"\
  73            f" AND {table}.osm_type = 'N'" + \
  74            " AND ST_Buffer(%s, reverse_place_diameter(%s)) && %s" % \
  75                tuple(map(lambda c: compiler.process(c, **kw), element.clauses))
  76
  77
  78 @compiles(IntersectsReverseDistance, 'sqlite') # type: ignore[no-untyped-call, misc]
  79 def sqlite_reverse_place_diameter(element: SaColumn,
  80                                   compiler: 'sa.Compiled', **kw: Any) -> str:
  81     geom1, rank, geom2 = list(element.clauses)
  82     table = element.tablename
  83
  84     return (f"{table}.rank_address between 4 and 25"\
  85             f" AND {table}.type != 'postcode'"\
  86             f" AND {table}.name is not null"\
  87             f" AND {table}.linked_place_id is null"\
  88             f" AND {table}.osm_type = 'N'"\
  89              " AND MbrIntersects(%s, ST_Expand(%s, 14.0 * exp(-0.2 * %s) - 0.03))"\
  90             f" AND {table}.place_id IN"\
  91              " (SELECT place_id FROM placex_place_node_areas"\
  92              "  WHERE ROWID IN (SELECT ROWID FROM SpatialIndex"\
  93              "  WHERE f_table_name = 'placex_place_node_areas'"\
  94              "  AND search_frame = %s))") % (
  95                 compiler.process(geom1, **kw),
  96                 compiler.process(geom2, **kw),
  97                 compiler.process(rank, **kw),
  98                 compiler.process(geom2, **kw))
  99
 100
 101 class IsBelowReverseDistance(sa.sql.functions.GenericFunction[bool]):
 102     type = sa.Boolean()
 103     name = 'IsBelowReverseDistance'
 104     inherit_cache = True
 105
 106
 107 @compiles(IsBelowReverseDistance) # type: ignore[no-untyped-call, misc]
 108 def default_is_below_reverse_distance(element: SaColumn,
 109                                       compiler: 'sa.Compiled', **kw: Any) -> str:
 110     dist, rank = list(element.clauses)
 111     return "%s < reverse_place_diameter(%s)" % (compiler.process(dist, **kw),
 112                                                 compiler.process(rank, **kw))
 113
 114
 115 @compiles(IsBelowReverseDistance, 'sqlite') # type: ignore[no-untyped-call, misc]
 116 def sqlite_is_below_reverse_distance(element: SaColumn,
 117                                      compiler: 'sa.Compiled', **kw: Any) -> str:
 118     dist, rank = list(element.clauses)
 119     return "%s < 14.0 * exp(-0.2 * %s) - 0.03" % (compiler.process(dist, **kw),
 120                                                   compiler.process(rank, **kw))
 121
 122
 123 def select_index_placex_geometry_reverse_lookupplacenode(table: str) -> 'sa.TextClause':
 124     """ Create an expression with the necessary conditions over a placex
 125         table that the index 'idx_placex_geometry_reverse_lookupPlaceNode'
 126         can be used.
 127     """
 128     return sa.text(f"{table}.rank_address between 4 and 25"
 129                    f" AND {table}.type != 'postcode'"
 130                    f" AND {table}.name is not null"
 131                    f" AND {table}.linked_place_id is null"
 132                    f" AND {table}.osm_type = 'N'")
 133
 134
 135 class CrosscheckNames(sa.sql.functions.GenericFunction[bool]):
 136     """ Check if in the given list of names in parameters 1 any of the names
 137         from the JSON array in parameter 2 are contained.
 138     """
 139     type = sa.Boolean()
 140     name = 'CrosscheckNames'
 141     inherit_cache = True
 142
 143 @compiles(CrosscheckNames) # type: ignore[no-untyped-call, misc]
 144 def compile_crosscheck_names(element: SaColumn,
 145                              compiler: 'sa.Compiled', **kw: Any) -> str:
 146     arg1, arg2 = list(element.clauses)
 147     return "coalesce(avals(%s) && ARRAY(SELECT * FROM json_array_elements_text(%s)), false)" % (
 148             compiler.process(arg1, **kw), compiler.process(arg2, **kw))
 149
 150
 151 @compiles(CrosscheckNames, 'sqlite') # type: ignore[no-untyped-call, misc]
 152 def compile_sqlite_crosscheck_names(element: SaColumn,
 153                                     compiler: 'sa.Compiled', **kw: Any) -> str:
 154     arg1, arg2 = list(element.clauses)
 155     return "EXISTS(SELECT *"\
 156            " FROM json_each(%s) as name, json_each(%s) as match_name"\
 157            " WHERE name.value = match_name.value)"\
 158            % (compiler.process(arg1, **kw), compiler.process(arg2, **kw))
 159
 160
 161 class JsonArrayEach(sa.sql.functions.GenericFunction[Any]):
 162     """ Return elements of a json array as a set.
 163     """
 164     name = 'JsonArrayEach'
 165     inherit_cache = True
 166
 167
 168 @compiles(JsonArrayEach) # type: ignore[no-untyped-call, misc]
 169 def default_json_array_each(element: SaColumn, compiler: 'sa.Compiled', **kw: Any) -> str:
 170     return "json_array_elements(%s)" % compiler.process(element.clauses, **kw)
 171
 172
 173 @compiles(JsonArrayEach, 'sqlite') # type: ignore[no-untyped-call, misc]
 174 def sqlite_json_array_each(element: SaColumn, compiler: 'sa.Compiled', **kw: Any) -> str:
 175     return "json_each(%s)" % compiler.process(element.clauses, **kw)
 176
 177
 178 class JsonHasKey(sa.sql.functions.GenericFunction[bool]):
 179     """ Return elements of a json array as a set.
 180     """
 181     type = sa.Boolean()
 182     name = 'JsonHasKey'
 183     inherit_cache = True
 184
 185
 186 @compiles(JsonHasKey) # type: ignore[no-untyped-call, misc]
 187 def compile_json_has_key(element: SaColumn,
 188                          compiler: 'sa.Compiled', **kw: Any) -> str:
 189     arg1, arg2 = list(element.clauses)
 190     return "%s->%s is not null" % (compiler.process(arg1, **kw),
 191                                    compiler.process(arg2, **kw))
 192
 193
 194 class Greatest(sa.sql.functions.GenericFunction[Any]):
 195     """ Function to compute maximum of all its input parameters.
 196     """
 197     name = 'greatest'
 198     inherit_cache = True
 199
 200
 201 @compiles(Greatest, 'sqlite') # type: ignore[no-untyped-call, misc]
 202 def sqlite_greatest(element: SaColumn, compiler: 'sa.Compiled', **kw: Any) -> str:
 203     return "max(%s)" % compiler.process(element.clauses, **kw)