safety-screen/Nominatim
1
0
Fork 0
forked from hans/Nominatim
Code Pull Requests Activity

indexer: allow batch processing of places

Browse Source 
Request and process multiple place_ids at once so that
Postgres can make better use of caching and there are less
transactions running.
This commit is contained in:
Sarah Hoffmann
2020-07-29 17:20:30 +02:00
parent a836ca8991
commit b2886426b7
2 changed files with 20 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
nominatim/indexer/progress.py
View File

@@ -21,7 +21,7 @@ class ProgressLogger(object):
self.total_places = total self.total_places = total
self.done_places = 0 self.done_places = 0
self.rank_start_time = datetime.now() self.rank_start_time = datetime.now()
self.next_info = 50 if log.isEnabledFor(logging.INFO) else total + 1 self.next_info = 100 if log.isEnabledFor(logging.INFO) else total + 1
def add(self, num=1): def add(self, num=1):
""" Mark `num` places as processed. Print a log message if the """ Mark `num` places as processed. Print a log message if the

32
nominatim/nominatim.py
View File

@@ -67,8 +67,9 @@ class RankRunner(object):
WHERE indexed_status > 0 and rank_search = {} WHERE indexed_status > 0 and rank_search = {}
ORDER BY geometry_sector""".format(self.rank) ORDER BY geometry_sector""".format(self.rank)
def sql_index_place(self): def sql_index_place(self, ids):
return "UPDATE placex SET indexed_status = 0 WHERE place_id = %s" return "UPDATE placex SET indexed_status = 0 WHERE place_id IN ({})"\
.format(','.join((str(i) for i in ids)))
class InterpolationRunner(object): class InterpolationRunner(object):
@@ -88,9 +89,10 @@ class InterpolationRunner(object):
WHERE indexed_status > 0 WHERE indexed_status > 0
ORDER BY geometry_sector""" ORDER BY geometry_sector"""
def sql_index_place(self): def sql_index_place(self, ids):
return """UPDATE location_property_osmline return """UPDATE location_property_osmline
SET indexed_status = 0 WHERE place_id = %s""" SET indexed_status = 0 WHERE place_id IN ({})"""\
.format(','.join((str(i) for i in ids)))
class DBConnection(object): class DBConnection(object):
@@ -199,13 +201,14 @@ class Indexer(object):
self.index(RankRunner(rank)) self.index(RankRunner(rank))
if self.maxrank == 30: if self.maxrank == 30:
self.index(InterpolationRunner()) self.index(InterpolationRunner(), 20)
self.index(RankRunner(self.maxrank)) self.index(RankRunner(self.maxrank), 20)
def index(self, obj): def index(self, obj, batch=1):
""" Index a single rank or table. `obj` describes the SQL to use """ Index a single rank or table. `obj` describes the SQL to use
for indexing. for indexing. `batch` describes the number of objects that
should be processed with a single SQL statement
""" """
log.warning("Starting {}".format(obj.name())) log.warning("Starting {}".format(obj.name()))
@@ -223,13 +226,16 @@ class Indexer(object):
cur = self.conn.cursor(name='places') cur = self.conn.cursor(name='places')
cur.execute(obj.sql_get_objects()) cur.execute(obj.sql_get_objects())
for place in cur: while True:
place_id = place[0] places = [p[0] for p in cur.fetchmany(batch)]
log.debug("Processing place {}".format(place_id)) if len(places) == 0:
break
log.debug("Processing places: {}".format(places))
thread = next(next_thread) thread = next(next_thread)
thread.perform(obj.sql_index_place(), (place_id,)) thread.perform(obj.sql_index_place(places))
progress.add() progress.add(len(places))
cur.close() cur.close()