nominatim/indexer/indexer.py

   1 """
   2 Main work horse for indexing (computing addresses) the database.
   3 """
   4 # pylint: disable=C0111
   5 import logging
   6 import select
   7
   8 import psycopg2
   9
  10 from .progress import ProgressLogger
  11 from ..db.async_connection import DBConnection
  12
  13 LOG = logging.getLogger()
  14
  15 class RankRunner:
  16     """ Returns SQL commands for indexing one rank within the placex table.
  17     """
  18
  19     def __init__(self, rank):
  20         self.rank = rank
  21
  22     def name(self):
  23         return "rank {}".format(self.rank)
  24
  25     def sql_count_objects(self):
  26         return """SELECT count(*) FROM placex
  27                   WHERE rank_address = {} and indexed_status > 0
  28                """.format(self.rank)
  29
  30     def sql_get_objects(self):
  31         return """SELECT place_id FROM placex
  32                   WHERE indexed_status > 0 and rank_address = {}
  33                   ORDER BY geometry_sector""".format(self.rank)
  34
  35     @staticmethod
  36     def sql_index_place(ids):
  37         return "UPDATE placex SET indexed_status = 0 WHERE place_id IN ({})"\
  38                .format(','.join((str(i) for i in ids)))
  39
  40
  41 class InterpolationRunner:
  42     """ Returns SQL commands for indexing the address interpolation table
  43         location_property_osmline.
  44     """
  45
  46     @staticmethod
  47     def name():
  48         return "interpolation lines (location_property_osmline)"
  49
  50     @staticmethod
  51     def sql_count_objects():
  52         return """SELECT count(*) FROM location_property_osmline
  53                   WHERE indexed_status > 0"""
  54
  55     @staticmethod
  56     def sql_get_objects():
  57         return """SELECT place_id FROM location_property_osmline
  58                   WHERE indexed_status > 0
  59                   ORDER BY geometry_sector"""
  60
  61     @staticmethod
  62     def sql_index_place(ids):
  63         return """UPDATE location_property_osmline
  64                   SET indexed_status = 0 WHERE place_id IN ({})"""\
  65                .format(','.join((str(i) for i in ids)))
  66
  67 class BoundaryRunner:
  68     """ Returns SQL commands for indexing the administrative boundaries
  69         of a certain rank.
  70     """
  71
  72     def __init__(self, rank):
  73         self.rank = rank
  74
  75     def name(self):
  76         return "boundaries rank {}".format(self.rank)
  77
  78     def sql_count_objects(self):
  79         return """SELECT count(*) FROM placex
  80                   WHERE indexed_status > 0
  81                     AND rank_search = {}
  82                     AND class = 'boundary' and type = 'administrative'""".format(self.rank)
  83
  84     def sql_get_objects(self):
  85         return """SELECT place_id FROM placex
  86                   WHERE indexed_status > 0 and rank_search = {}
  87                         and class = 'boundary' and type = 'administrative'
  88                   ORDER BY partition, admin_level""".format(self.rank)
  89
  90     @staticmethod
  91     def sql_index_place(ids):
  92         return "UPDATE placex SET indexed_status = 0 WHERE place_id IN ({})"\
  93                .format(','.join((str(i) for i in ids)))
  94
  95 class Indexer:
  96     """ Main indexing routine.
  97     """
  98
  99     def __init__(self, dsn, num_threads):
 100         self.conn = psycopg2.connect(dsn)
 101         self.threads = [DBConnection(dsn) for _ in range(num_threads)]
 102
 103     def index_boundaries(self, minrank, maxrank):
 104         LOG.warning("Starting indexing boundaries using %s threads",
 105                     len(self.threads))
 106
 107         for rank in range(max(minrank, 4), min(maxrank, 26)):
 108             self.index(BoundaryRunner(rank))
 109
 110     def index_by_rank(self, minrank, maxrank):
 111         """ Run classic indexing by rank.
 112         """
 113         maxrank = min(maxrank, 30)
 114         LOG.warning("Starting indexing rank (%i to %i) using %i threads",
 115                     minrank, maxrank, len(self.threads))
 116
 117         for rank in range(max(1, minrank), maxrank):
 118             self.index(RankRunner(rank))
 119
 120         if maxrank == 30:
 121             self.index(RankRunner(0))
 122             self.index(InterpolationRunner(), 20)
 123             self.index(RankRunner(30), 20)
 124         else:
 125             self.index(RankRunner(maxrank))
 126
 127     def update_status_table(self):
 128         """ Update the status in the status table to 'indexed'.
 129         """
 130         with self.conn.cursor() as cur:
 131             cur.execute('UPDATE import_status SET indexed = true')
 132         self.conn.commit()
 133
 134     def index(self, obj, batch=1):
 135         """ Index a single rank or table. `obj` describes the SQL to use
 136             for indexing. `batch` describes the number of objects that
 137             should be processed with a single SQL statement
 138         """
 139         LOG.warning("Starting %s (using batch size %s)", obj.name(), batch)
 140
 141         cur = self.conn.cursor()
 142         cur.execute(obj.sql_count_objects())
 143
 144         total_tuples = cur.fetchone()[0]
 145         LOG.debug("Total number of rows: %i", total_tuples)
 146
 147         cur.close()
 148
 149         progress = ProgressLogger(obj.name(), total_tuples)
 150
 151         if total_tuples > 0:
 152             cur = self.conn.cursor(name='places')
 153             cur.execute(obj.sql_get_objects())
 154
 155             next_thread = self.find_free_thread()
 156             while True:
 157                 places = [p[0] for p in cur.fetchmany(batch)]
 158                 if not places:
 159                     break
 160
 161                 LOG.debug("Processing places: %s", str(places))
 162                 thread = next(next_thread)
 163
 164                 thread.perform(obj.sql_index_place(places))
 165                 progress.add(len(places))
 166
 167             cur.close()
 168
 169             for thread in self.threads:
 170                 thread.wait()
 171
 172         progress.done()
 173
 174     def find_free_thread(self):
 175         """ Generator that returns the next connection that is free for
 176             sending a query.
 177         """
 178         ready = self.threads
 179         command_stat = 0
 180
 181         while True:
 182             for thread in ready:
 183                 if thread.is_done():
 184                     command_stat += 1
 185                     yield thread
 186
 187             # refresh the connections occasionaly to avoid potential
 188             # memory leaks in Postgresql.
 189             if command_stat > 100000:
 190                 for thread in self.threads:
 191                     while not thread.is_done():
 192                         thread.wait()
 193                     thread.connect()
 194                 command_stat = 0
 195                 ready = self.threads
 196             else:
 197                 ready, _, _ = select.select(self.threads, [], [])
 198
 199         assert False, "Unreachable code"