nominatim/indexer/runners.py

   1 """
   2 Mix-ins that provide the actual commands for the indexer for various indexing
   3 tasks.
   4 """
   5 import functools
   6
   7 import psycopg2.extras
   8 from psycopg2 import sql as pysql
   9
  10 # pylint: disable=C0111
  11
  12 def _mk_valuelist(template, num):
  13     return pysql.SQL(',').join([pysql.SQL(template)] * num)
  14
  15 class AbstractPlacexRunner:
  16     """ Returns SQL commands for indexing of the placex table.
  17     """
  18     SELECT_SQL = pysql.SQL('SELECT place_id FROM placex ')
  19     UPDATE_LINE = "(%s, %s::hstore, %s::hstore, %s::int, %s::jsonb)"
  20
  21     def __init__(self, rank, analyzer):
  22         self.rank = rank
  23         self.analyzer = analyzer
  24
  25
  26     @staticmethod
  27     @functools.lru_cache(maxsize=1)
  28     def _index_sql(num_places):
  29         return pysql.SQL(
  30             """ UPDATE placex
  31                 SET indexed_status = 0, address = v.addr, token_info = v.ti,
  32                     name = v.name, linked_place_id = v.linked_place_id
  33                 FROM (VALUES {}) as v(id, name, addr, linked_place_id, ti)
  34                 WHERE place_id = v.id
  35             """).format(_mk_valuelist(AbstractPlacexRunner.UPDATE_LINE, num_places))
  36
  37
  38     @staticmethod
  39     def get_place_details(worker, ids):
  40         worker.perform("""SELECT place_id, (placex_prepare_update(placex)).*
  41                           FROM placex WHERE place_id IN %s""",
  42                        (tuple((p[0] for p in ids)), ))
  43
  44
  45     def index_places(self, worker, places):
  46         values = []
  47         for place in places:
  48             for field in ('place_id', 'name', 'address', 'linked_place_id'):
  49                 values.append(place[field])
  50             values.append(psycopg2.extras.Json(self.analyzer.process_place(place)))
  51
  52         worker.perform(self._index_sql(len(places)), values)
  53
  54
  55 class RankRunner(AbstractPlacexRunner):
  56     """ Returns SQL commands for indexing one rank within the placex table.
  57     """
  58
  59     def name(self):
  60         return "rank {}".format(self.rank)
  61
  62     def sql_count_objects(self):
  63         return pysql.SQL("""SELECT count(*) FROM placex
  64                             WHERE rank_address = {} and indexed_status > 0
  65                          """).format(pysql.Literal(self.rank))
  66
  67     def sql_get_objects(self):
  68         return self.SELECT_SQL + pysql.SQL(
  69             """WHERE indexed_status > 0 and rank_address = {}
  70                ORDER BY geometry_sector
  71             """).format(pysql.Literal(self.rank))
  72
  73
  74 class BoundaryRunner(AbstractPlacexRunner):
  75     """ Returns SQL commands for indexing the administrative boundaries
  76         of a certain rank.
  77     """
  78
  79     def name(self):
  80         return "boundaries rank {}".format(self.rank)
  81
  82     def sql_count_objects(self):
  83         return pysql.SQL("""SELECT count(*) FROM placex
  84                             WHERE indexed_status > 0
  85                               AND rank_search = {}
  86                               AND class = 'boundary' and type = 'administrative'
  87                          """).format(pysql.Literal(self.rank))
  88
  89     def sql_get_objects(self):
  90         return self.SELECT_SQL + pysql.SQL(
  91             """WHERE indexed_status > 0 and rank_search = {}
  92                      and class = 'boundary' and type = 'administrative'
  93                ORDER BY partition, admin_level
  94             """).format(pysql.Literal(self.rank))
  95
  96
  97 class InterpolationRunner:
  98     """ Returns SQL commands for indexing the address interpolation table
  99         location_property_osmline.
 100     """
 101
 102     def __init__(self, analyzer):
 103         self.analyzer = analyzer
 104
 105
 106     @staticmethod
 107     def name():
 108         return "interpolation lines (location_property_osmline)"
 109
 110     @staticmethod
 111     def sql_count_objects():
 112         return """SELECT count(*) FROM location_property_osmline
 113                   WHERE indexed_status > 0"""
 114
 115     @staticmethod
 116     def sql_get_objects():
 117         return """SELECT place_id
 118                   FROM location_property_osmline
 119                   WHERE indexed_status > 0
 120                   ORDER BY geometry_sector"""
 121
 122
 123     @staticmethod
 124     def get_place_details(worker, ids):
 125         worker.perform("""SELECT place_id, get_interpolation_address(address, osm_id) as address
 126                           FROM location_property_osmline WHERE place_id IN %s""",
 127                        (tuple((p[0] for p in ids)), ))
 128
 129
 130     @staticmethod
 131     @functools.lru_cache(maxsize=1)
 132     def _index_sql(num_places):
 133         return pysql.SQL("""UPDATE location_property_osmline
 134                             SET indexed_status = 0, address = v.addr, token_info = v.ti
 135                             FROM (VALUES {}) as v(id, addr, ti)
 136                             WHERE place_id = v.id
 137                          """).format(_mk_valuelist("(%s, %s::hstore, %s::jsonb)", num_places))
 138
 139
 140     def index_places(self, worker, places):
 141         values = []
 142         for place in places:
 143             values.extend((place[x] for x in ('place_id', 'address')))
 144             values.append(psycopg2.extras.Json(self.analyzer.process_place(place)))
 145
 146         worker.perform(self._index_sql(len(places)), values)
 147
 148
 149
 150 class PostcodeRunner:
 151     """ Provides the SQL commands for indexing the location_postcode table.
 152     """
 153
 154     @staticmethod
 155     def name():
 156         return "postcodes (location_postcode)"
 157
 158     @staticmethod
 159     def sql_count_objects():
 160         return 'SELECT count(*) FROM location_postcode WHERE indexed_status > 0'
 161
 162     @staticmethod
 163     def sql_get_objects():
 164         return """SELECT place_id FROM location_postcode
 165                   WHERE indexed_status > 0
 166                   ORDER BY country_code, postcode"""
 167
 168     @staticmethod
 169     def index_places(worker, ids):
 170         worker.perform(pysql.SQL("""UPDATE location_postcode SET indexed_status = 0
 171                                     WHERE place_id IN ({})""")
 172                        .format(pysql.SQL(',').join((pysql.Literal(i[0]) for i in ids))))