X-Git-Url: https://git.openstreetmap.org./nominatim.git/blobdiff_plain/696c50459fc3f0a905211787e509a13b4bb96ea4..83054af46fb21ea40c0bebb2c8a528937e5a49a0:/nominatim/indexer/runners.py diff --git a/nominatim/indexer/runners.py b/nominatim/indexer/runners.py index 3c853cd0..bbadd282 100644 --- a/nominatim/indexer/runners.py +++ b/nominatim/indexer/runners.py @@ -1,64 +1,123 @@ +# SPDX-License-Identifier: GPL-2.0-only +# +# This file is part of Nominatim. (https://nominatim.org) +# +# Copyright (C) 2022 by the Nominatim developer community. +# For a full list of authors see the git log. """ Mix-ins that provide the actual commands for the indexer for various indexing tasks. """ +from typing import Any, List +import functools + +from psycopg2 import sql as pysql +import psycopg2.extras + +from nominatim.data.place_info import PlaceInfo +from nominatim.tokenizer.base import AbstractAnalyzer +from nominatim.db.async_connection import DBConnection +from nominatim.typing import Query, DictCursorResult, DictCursorResults, Protocol + # pylint: disable=C0111 -class RankRunner: - """ Returns SQL commands for indexing one rank within the placex table. +def _mk_valuelist(template: str, num: int) -> pysql.Composed: + return pysql.SQL(',').join([pysql.SQL(template)] * num) + +def _analyze_place(place: DictCursorResult, analyzer: AbstractAnalyzer) -> psycopg2.extras.Json: + return psycopg2.extras.Json(analyzer.process_place(PlaceInfo(place))) + + +class Runner(Protocol): + def name(self) -> str: ... + def sql_count_objects(self) -> Query: ... + def sql_get_objects(self) -> Query: ... + def get_place_details(self, worker: DBConnection, + ids: DictCursorResults) -> DictCursorResults: ... + def index_places(self, worker: DBConnection, places: DictCursorResults) -> None: ... + + +class AbstractPlacexRunner: + """ Returns SQL commands for indexing of the placex table. """ + SELECT_SQL = pysql.SQL('SELECT place_id FROM placex ') + UPDATE_LINE = "(%s, %s::hstore, %s::hstore, %s::int, %s::jsonb)" - def __init__(self, rank): + def __init__(self, rank: int, analyzer: AbstractAnalyzer) -> None: self.rank = rank + self.analyzer = analyzer - def name(self): - return "rank {}".format(self.rank) - def sql_count_objects(self): - return """SELECT count(*) FROM placex - WHERE rank_address = {} and indexed_status > 0 - """.format(self.rank) + @functools.lru_cache(maxsize=1) + def _index_sql(self, num_places: int) -> pysql.Composed: + return pysql.SQL( + """ UPDATE placex + SET indexed_status = 0, address = v.addr, token_info = v.ti, + name = v.name, linked_place_id = v.linked_place_id + FROM (VALUES {}) as v(id, name, addr, linked_place_id, ti) + WHERE place_id = v.id + """).format(_mk_valuelist(AbstractPlacexRunner.UPDATE_LINE, num_places)) - def sql_get_objects(self): - return """SELECT place_id FROM placex - WHERE indexed_status > 0 and rank_address = {} - ORDER BY geometry_sector""".format(self.rank) - @staticmethod - def sql_index_place(ids): - return "UPDATE placex SET indexed_status = 0 WHERE place_id IN ({})"\ - .format(','.join((str(i) for i in ids))) + def get_place_details(self, worker: DBConnection, ids: DictCursorResults) -> DictCursorResults: + worker.perform("""SELECT place_id, extra.* + FROM placex, LATERAL placex_indexing_prepare(placex) as extra + WHERE place_id IN %s""", + (tuple((p[0] for p in ids)), )) + return [] -class BoundaryRunner: - """ Returns SQL commands for indexing the administrative boundaries - of a certain rank. + + def index_places(self, worker: DBConnection, places: DictCursorResults) -> None: + values: List[Any] = [] + for place in places: + for field in ('place_id', 'name', 'address', 'linked_place_id'): + values.append(place[field]) + values.append(_analyze_place(place, self.analyzer)) + + worker.perform(self._index_sql(len(places)), values) + + +class RankRunner(AbstractPlacexRunner): + """ Returns SQL commands for indexing one rank within the placex table. """ - def __init__(self, rank): - self.rank = rank + def name(self) -> str: + return f"rank {self.rank}" - def name(self): - return "boundaries rank {}".format(self.rank) + def sql_count_objects(self) -> pysql.Composed: + return pysql.SQL("""SELECT count(*) FROM placex + WHERE rank_address = {} and indexed_status > 0 + """).format(pysql.Literal(self.rank)) - def sql_count_objects(self): - return """SELECT count(*) FROM placex - WHERE indexed_status > 0 - AND rank_search = {} - AND class = 'boundary' and type = 'administrative' - """.format(self.rank) + def sql_get_objects(self) -> pysql.Composed: + return self.SELECT_SQL + pysql.SQL( + """WHERE indexed_status > 0 and rank_address = {} + ORDER BY geometry_sector + """).format(pysql.Literal(self.rank)) - def sql_get_objects(self): - return """SELECT place_id FROM placex - WHERE indexed_status > 0 and rank_search = {} - and class = 'boundary' and type = 'administrative' - ORDER BY partition, admin_level - """.format(self.rank) - @staticmethod - def sql_index_place(ids): - return "UPDATE placex SET indexed_status = 0 WHERE place_id IN ({})"\ - .format(','.join((str(i) for i in ids))) +class BoundaryRunner(AbstractPlacexRunner): + """ Returns SQL commands for indexing the administrative boundaries + of a certain rank. + """ + + def name(self) -> str: + return f"boundaries rank {self.rank}" + + def sql_count_objects(self) -> pysql.Composed: + return pysql.SQL("""SELECT count(*) FROM placex + WHERE indexed_status > 0 + AND rank_search = {} + AND class = 'boundary' and type = 'administrative' + """).format(pysql.Literal(self.rank)) + + def sql_get_objects(self) -> pysql.Composed: + return self.SELECT_SQL + pysql.SQL( + """WHERE indexed_status > 0 and rank_search = {} + and class = 'boundary' and type = 'administrative' + ORDER BY partition, admin_level + """).format(pysql.Literal(self.rank)) class InterpolationRunner: @@ -66,48 +125,72 @@ class InterpolationRunner: location_property_osmline. """ - @staticmethod - def name(): + def __init__(self, analyzer: AbstractAnalyzer) -> None: + self.analyzer = analyzer + + + def name(self) -> str: return "interpolation lines (location_property_osmline)" - @staticmethod - def sql_count_objects(): + def sql_count_objects(self) -> str: return """SELECT count(*) FROM location_property_osmline WHERE indexed_status > 0""" - @staticmethod - def sql_get_objects(): - return """SELECT place_id FROM location_property_osmline + def sql_get_objects(self) -> str: + return """SELECT place_id + FROM location_property_osmline WHERE indexed_status > 0 ORDER BY geometry_sector""" - @staticmethod - def sql_index_place(ids): - return """UPDATE location_property_osmline - SET indexed_status = 0 WHERE place_id IN ({}) - """.format(','.join((str(i) for i in ids))) + def get_place_details(self, worker: DBConnection, ids: DictCursorResults) -> DictCursorResults: + worker.perform("""SELECT place_id, get_interpolation_address(address, osm_id) as address + FROM location_property_osmline WHERE place_id IN %s""", + (tuple((p[0] for p in ids)), )) + return [] + + + @functools.lru_cache(maxsize=1) + def _index_sql(self, num_places: int) -> pysql.Composed: + return pysql.SQL("""UPDATE location_property_osmline + SET indexed_status = 0, address = v.addr, token_info = v.ti + FROM (VALUES {}) as v(id, addr, ti) + WHERE place_id = v.id + """).format(_mk_valuelist("(%s, %s::hstore, %s::jsonb)", num_places)) -class PostcodeRunner: + + def index_places(self, worker: DBConnection, places: DictCursorResults) -> None: + values: List[Any] = [] + for place in places: + values.extend((place[x] for x in ('place_id', 'address'))) + values.append(_analyze_place(place, self.analyzer)) + + worker.perform(self._index_sql(len(places)), values) + + + +class PostcodeRunner(Runner): """ Provides the SQL commands for indexing the location_postcode table. """ - @staticmethod - def name(): + def name(self) -> str: return "postcodes (location_postcode)" - @staticmethod - def sql_count_objects(): + + def sql_count_objects(self) -> str: return 'SELECT count(*) FROM location_postcode WHERE indexed_status > 0' - @staticmethod - def sql_get_objects(): + + def sql_get_objects(self) -> str: return """SELECT place_id FROM location_postcode WHERE indexed_status > 0 ORDER BY country_code, postcode""" - @staticmethod - def sql_index_place(ids): - return """UPDATE location_postcode SET indexed_status = 0 - WHERE place_id IN ({}) - """.format(','.join((str(i) for i in ids))) + + def get_place_details(self, worker: DBConnection, ids: DictCursorResults) -> DictCursorResults: + return ids + + def index_places(self, worker: DBConnection, places: DictCursorResults) -> None: + worker.perform(pysql.SQL("""UPDATE location_postcode SET indexed_status = 0 + WHERE place_id IN ({})""") + .format(pysql.SQL(',').join((pysql.Literal(i[0]) for i in places))))