"""
Main work horse for indexing (computing addresses) the database.
"""
-# pylint: disable=C0111
import logging
import select
-import psycopg2
-
-from .progress import ProgressLogger
-from ..db.async_connection import DBConnection
+from nominatim.indexer.progress import ProgressLogger
+from nominatim.indexer import runners
+from nominatim.db.async_connection import DBConnection
+from nominatim.db.connection import connect
LOG = logging.getLogger()
-class RankRunner:
- """ Returns SQL commands for indexing one rank within the placex table.
- """
-
- def __init__(self, rank):
- self.rank = rank
-
- 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)
-
- 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)))
-
-
-class InterpolationRunner:
- """ Returns SQL commands for indexing the address interpolation table
- location_property_osmline.
- """
-
- @staticmethod
- def name():
- return "interpolation lines (location_property_osmline)"
-
- @staticmethod
- def sql_count_objects():
- return """SELECT count(*) FROM location_property_osmline
- WHERE indexed_status > 0"""
-
- @staticmethod
- def sql_get_objects():
- 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)))
-
-class BoundaryRunner:
- """ Returns SQL commands for indexing the administrative boundaries
- of a certain rank.
- """
-
- def __init__(self, rank):
- self.rank = rank
-
- def name(self):
- return "boundaries rank {}".format(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):
- 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 PostcodeRunner:
- """ Provides the SQL commands for indexing the location_postcode table.
- """
-
- @staticmethod
- def name():
- return "postcodes (location_postcode)"
-
- @staticmethod
- def sql_count_objects():
- return 'SELECT count(*) FROM location_postcode WHERE indexed_status > 0'
-
- @staticmethod
- def sql_get_objects():
- 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)))
class Indexer:
""" Main indexing routine.
def __init__(self, dsn, num_threads):
self.dsn = dsn
self.num_threads = num_threads
- self.conn = None
self.threads = []
def _setup_connections(self):
- self.conn = psycopg2.connect(self.dsn)
self.threads = [DBConnection(self.dsn) for _ in range(self.num_threads)]
def _close_connections(self):
- if self.conn:
- self.conn.close()
- self.conn = None
-
for thread in self.threads:
thread.close()
- threads = []
+ self.threads = []
def index_full(self, analyse=True):
database will be analysed at the appropriate places to
ensure that database statistics are updated.
"""
- conn = psycopg2.connect(self.dsn)
+ with connect(self.dsn) as conn:
+ conn.autocommit = True
+
+ if analyse:
+ def _analyse():
+ with conn.cursor() as cur:
+ cur.execute('ANALYSE')
+ else:
+ def _analyse():
+ pass
- try:
self.index_by_rank(0, 4)
- self._analyse_db_if(conn, analyse)
+ _analyse()
self.index_boundaries(0, 30)
- self._analyse_db_if(conn, analyse)
+ _analyse()
self.index_by_rank(5, 25)
- self._analyse_db_if(conn, analyse)
+ _analyse()
self.index_by_rank(26, 30)
- self._analyse_db_if(conn, analyse)
+ _analyse()
self.index_postcodes()
- self._analyse_db_if(conn, analyse)
- finally:
- conn.close()
+ _analyse()
- def _analyse_db_if(self, conn, condition):
- if condition:
- with conn.cursor() as cur:
- cur.execute('ANALYSE')
def index_boundaries(self, minrank, maxrank):
""" Index only administrative boundaries within the given rank range.
try:
for rank in range(max(minrank, 4), min(maxrank, 26)):
- self.index(BoundaryRunner(rank))
+ self._index(runners.BoundaryRunner(rank))
finally:
self._close_connections()
try:
for rank in range(max(1, minrank), maxrank):
- self.index(RankRunner(rank))
+ self._index(runners.RankRunner(rank))
if maxrank == 30:
- self.index(RankRunner(0))
- self.index(InterpolationRunner(), 20)
- self.index(RankRunner(30), 20)
+ self._index(runners.RankRunner(0))
+ self._index(runners.InterpolationRunner(), 20)
+ self._index(runners.RankRunner(30), 20)
else:
- self.index(RankRunner(maxrank))
+ self._index(runners.RankRunner(maxrank))
finally:
self._close_connections()
self._setup_connections()
try:
- self.index(PostcodeRunner(), 20)
+ self._index(runners.PostcodeRunner(), 20)
finally:
self._close_connections()
def update_status_table(self):
""" Update the status in the status table to 'indexed'.
"""
- conn = psycopg2.connect(self.dsn)
-
- try:
+ with connect(self.dsn) as conn:
with conn.cursor() as cur:
cur.execute('UPDATE import_status SET indexed = true')
conn.commit()
- finally:
- conn.close()
- def index(self, obj, batch=1):
- """ Index a single rank or table. `obj` describes the SQL to use
+ def _index(self, runner, batch=1):
+ """ Index a single rank or table. `runner` describes the SQL to use
for indexing. `batch` describes the number of objects that
should be processed with a single SQL statement
"""
- LOG.warning("Starting %s (using batch size %s)", obj.name(), batch)
-
- cur = self.conn.cursor()
- cur.execute(obj.sql_count_objects())
+ LOG.warning("Starting %s (using batch size %s)", runner.name(), batch)
- total_tuples = cur.fetchone()[0]
- LOG.debug("Total number of rows: %i", total_tuples)
+ with connect(self.dsn) as conn:
+ with conn.cursor() as cur:
+ total_tuples = cur.scalar(runner.sql_count_objects())
+ LOG.debug("Total number of rows: %i", total_tuples)
- cur.close()
+ conn.commit()
- progress = ProgressLogger(obj.name(), total_tuples)
+ progress = ProgressLogger(runner.name(), total_tuples)
- if total_tuples > 0:
- cur = self.conn.cursor(name='places')
- cur.execute(obj.sql_get_objects())
+ if total_tuples > 0:
+ with conn.cursor(name='places') as cur:
+ cur.execute(runner.sql_get_objects())
- next_thread = self.find_free_thread()
- while True:
- places = [p[0] for p in cur.fetchmany(batch)]
- if not places:
- break
+ next_thread = self.find_free_thread()
+ while True:
+ places = [p[0] for p in cur.fetchmany(batch)]
+ if not places:
+ break
- LOG.debug("Processing places: %s", str(places))
- thread = next(next_thread)
+ LOG.debug("Processing places: %s", str(places))
+ thread = next(next_thread)
- thread.perform(obj.sql_index_place(places))
- progress.add(len(places))
+ thread.perform(runner.sql_index_place(places))
+ progress.add(len(places))
- cur.close()
+ conn.commit()
- for thread in self.threads:
- thread.wait()
+ for thread in self.threads:
+ thread.wait()
progress.done()
projects / nominatim.git / blobdiff