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