hans/Nominatim
1
0
Fork 1
mirror of https://github.com/osm-search/Nominatim.git synced 2026-03-11 21:34:06 +00:00
Code Issues Packages Projects Releases Wiki Activity

indexer: fetch ids in batches

Browse Source
This commit is contained in:
Sarah Hoffmann
2021-04-29 21:57:43 +02:00
parent fc995ea6b9
commit 602728895e
1 changed files with 39 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
nominatim/indexer/indexer.py
View File

@@ -183,6 +183,9 @@ class Indexer:
total_tuples = cur.scalar(runner.sql_count_objects()) total_tuples = cur.scalar(runner.sql_count_objects())
LOG.debug("Total number of rows: %i", total_tuples) LOG.debug("Total number of rows: %i", total_tuples)
hstore_oid = cur.scalar("SELECT 'hstore'::regtype::oid")
hstore_array_oid = cur.scalar("SELECT 'hstore[]'::regtype::oid")
conn.commit() conn.commit()
progress = ProgressLogger(runner.name(), total_tuples) progress = ProgressLogger(runner.name(), total_tuples)
@@ -191,20 +194,49 @@ class Indexer:
with conn.cursor(name='places') as cur: with conn.cursor(name='places') as cur:
cur.execute(runner.sql_get_objects()) cur.execute(runner.sql_get_objects())
fetcher = DBConnection(self.dsn)
psycopg2.extras.register_hstore(fetcher.conn,
oid=hstore_oid,
array_oid=hstore_array_oid)
with WorkerPool(self.dsn, self.num_threads) as pool: with WorkerPool(self.dsn, self.num_threads) as pool:
while True: places = self._fetch_next_batch(cur, fetcher, runner)
places = cur.fetchmany(batch) while places is not None:
if not places: if not places:
break fetcher.wait()
places = fetcher.cursor.fetchall()
LOG.debug("Processing places: %s", str(places)) # asynchronously get the next batch
worker = pool.next_free_worker() next_places = self._fetch_next_batch(cur, fetcher, runner)
runner.index_places(worker, places) # And insert the curent batch
progress.add(len(places)) for idx in range(0, len(places), batch):
worker = pool.next_free_worker()
part = places[idx:idx+batch]
LOG.debug("Processing places: %s", str(part))
runner.index_places(worker, part)
progress.add(len(part))
places = next_places
pool.finish_all() pool.finish_all()
fetcher.wait()
fetcher.close()
conn.commit() conn.commit()
progress.done() progress.done()
def _fetch_next_batch(self, cur, fetcher, runner):
ids = cur.fetchmany(1000)
if not ids:
return None
if not hasattr(runner, 'get_place_details'):
return ids
runner.get_place_details(fetcher, ids)
return []