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