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=abstract-method,missing-function-docstring,consider-using-f-string
  19
  20 def select_index_placex_geometry_reverse_lookuppolygon(table: str) -> 'sa.TextClause':
  21     """ Create an expression with the necessary conditions over a placex
  22         table that the index 'idx_placex_geometry_reverse_lookupPolygon'
  23         can be used.
  24     """
  25     return sa.text(f"ST_GeometryType({table}.geometry) in ('ST_Polygon', 'ST_MultiPolygon')"
  26                    f" AND {table}.rank_address between 4 and 25"
  27                    f" AND {table}.type != 'postcode'"
  28                    f" AND {table}.name is not null"
  29                    f" AND {table}.indexed_status = 0"
  30                    f" AND {table}.linked_place_id is null")
  31
  32 def select_index_placex_geometry_reverse_lookupplacenode(table: str) -> 'sa.TextClause':
  33     """ Create an expression with the necessary conditions over a placex
  34         table that the index 'idx_placex_geometry_reverse_lookupPlaceNode'
  35         can be used.
  36     """
  37     return sa.text(f"{table}.rank_address between 4 and 25"
  38                    f" AND {table}.type != 'postcode'"
  39                    f" AND {table}.name is not null"
  40                    f" AND {table}.linked_place_id is null"
  41                    f" AND {table}.osm_type = 'N'")
  42
  43
  44 class CrosscheckNames(sa.sql.functions.GenericFunction[bool]):
  45     """ Check if in the given list of names in parameters 1 any of the names
  46         from the JSON array in parameter 2 are contained.
  47     """
  48     type = sa.Boolean()
  49     name = 'CrosscheckNames'
  50     inherit_cache = True
  51
  52 @compiles(CrosscheckNames) # type: ignore[no-untyped-call, misc]
  53 def compile_crosscheck_names(element: SaColumn,
  54                              compiler: 'sa.Compiled', **kw: Any) -> str:
  55     arg1, arg2 = list(element.clauses)
  56     return "coalesce(avals(%s) && ARRAY(SELECT * FROM json_array_elements_text(%s)), false)" % (
  57             compiler.process(arg1, **kw), compiler.process(arg2, **kw))
  58
  59
  60 @compiles(CrosscheckNames, 'sqlite') # type: ignore[no-untyped-call, misc]
  61 def compile_sqlite_crosscheck_names(element: SaColumn,
  62                                     compiler: 'sa.Compiled', **kw: Any) -> str:
  63     arg1, arg2 = list(element.clauses)
  64     return "EXISTS(SELECT *"\
  65            " FROM json_each(%s) as name, json_each(%s) as match_name"\
  66            " WHERE name.value = match_name.value)"\
  67            % (compiler.process(arg1, **kw), compiler.process(arg2, **kw))
  68
  69
  70 class JsonArrayEach(sa.sql.functions.GenericFunction[Any]):
  71     """ Return elements of a json array as a set.
  72     """
  73     name = 'JsonArrayEach'
  74     inherit_cache = True
  75
  76
  77 @compiles(JsonArrayEach) # type: ignore[no-untyped-call, misc]
  78 def default_json_array_each(element: SaColumn, compiler: 'sa.Compiled', **kw: Any) -> str:
  79     return "json_array_elements(%s)" % compiler.process(element.clauses, **kw)
  80
  81
  82 @compiles(JsonArrayEach, 'sqlite') # type: ignore[no-untyped-call, misc]
  83 def sqlite_json_array_each(element: SaColumn, compiler: 'sa.Compiled', **kw: Any) -> str:
  84     return "json_each(%s)" % compiler.process(element.clauses, **kw)
  85
  86
  87 class Greatest(sa.sql.functions.GenericFunction[Any]):
  88     """ Function to compute maximum of all its input parameters.
  89     """
  90     name = 'greatest'
  91     inherit_cache = True
  92
  93
  94 @compiles(Greatest, 'sqlite') # type: ignore[no-untyped-call, misc]
  95 def sqlite_greatest(element: SaColumn, compiler: 'sa.Compiled', **kw: Any) -> str:
  96     return "max(%s)" % compiler.process(element.clauses, **kw)