+# SPDX-License-Identifier: GPL-2.0-only
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2022 by the Nominatim developer community.
+# For a full list of authors see the git log.
"""
Main work horse for indexing (computing addresses) the database.
"""
+from typing import Optional, Any, cast
import logging
-import select
import time
import psycopg2.extras
+from nominatim.tokenizer.base import AbstractTokenizer
from nominatim.indexer.progress import ProgressLogger
from nominatim.indexer import runners
-from nominatim.db.async_connection import DBConnection
-from nominatim.db.connection import connect
+from nominatim.db.async_connection import DBConnection, WorkerPool
+from nominatim.db.connection import connect, Connection, Cursor
+from nominatim.typing import DictCursorResults
LOG = logging.getLogger()
-class WorkerPool:
- """ A pool of asynchronous database connections.
- The pool may be used as a context manager.
+class PlaceFetcher:
+ """ Asynchronous connection that fetches place details for processing.
"""
- REOPEN_CONNECTIONS_AFTER = 100000
+ def __init__(self, dsn: str, setup_conn: Connection) -> None:
+ self.wait_time = 0.0
+ self.current_ids: Optional[DictCursorResults] = None
+ self.conn: Optional[DBConnection] = DBConnection(dsn,
+ cursor_factory=psycopg2.extras.DictCursor)
+
+ with setup_conn.cursor() as cur:
+ # need to fetch those manually because register_hstore cannot
+ # fetch them on an asynchronous connection below.
+ hstore_oid = cur.scalar("SELECT 'hstore'::regtype::oid")
+ hstore_array_oid = cur.scalar("SELECT 'hstore[]'::regtype::oid")
+
+ psycopg2.extras.register_hstore(self.conn.conn, oid=hstore_oid,
+ array_oid=hstore_array_oid)
+
+ def close(self) -> None:
+ """ Close the underlying asynchronous connection.
+ """
+ if self.conn:
+ self.conn.close()
+ self.conn = None
- def __init__(self, dsn, pool_size):
- self.threads = [DBConnection(dsn) for _ in range(pool_size)]
- self.free_workers = self._yield_free_worker()
+ def fetch_next_batch(self, cur: Cursor, runner: runners.Runner) -> bool:
+ """ Send a request for the next batch of places.
+ If details for the places are required, they will be fetched
+ asynchronously.
- def finish_all(self):
- """ Wait for all connection to finish.
+ Returns true if there is still data available.
"""
- for thread in self.threads:
- while not thread.is_done():
- thread.wait()
+ ids = cast(Optional[DictCursorResults], cur.fetchmany(100))
- self.free_workers = self._yield_free_worker()
+ if not ids:
+ self.current_ids = None
+ return False
- def close(self):
- """ Close all connections and clear the pool.
- """
- for thread in self.threads:
- thread.close()
- self.threads = []
- self.free_workers = None
+ assert self.conn is not None
+ self.current_ids = runner.get_place_details(self.conn, ids)
+ return True
- def next_free_worker(self):
- """ Get the next free connection.
+ def get_batch(self) -> DictCursorResults:
+ """ Get the next batch of data, previously requested with
+ `fetch_next_batch`.
"""
- return next(self.free_workers)
-
-
- def _yield_free_worker(self):
- ready = self.threads
- command_stat = 0
- while True:
- for thread in ready:
- if thread.is_done():
- command_stat += 1
- yield thread
-
- if command_stat > self.REOPEN_CONNECTIONS_AFTER:
- for thread in self.threads:
- while not thread.is_done():
- thread.wait()
- thread.connect()
- ready = self.threads
- command_stat = 0
- else:
- _, ready, _ = select.select([], self.threads, [])
-
-
- def __enter__(self):
+ assert self.conn is not None
+ assert self.conn.cursor is not None
+
+ if self.current_ids is not None and not self.current_ids:
+ tstart = time.time()
+ self.conn.wait()
+ self.wait_time += time.time() - tstart
+ self.current_ids = cast(Optional[DictCursorResults],
+ self.conn.cursor.fetchall())
+
+ return self.current_ids if self.current_ids is not None else []
+
+ def __enter__(self) -> 'PlaceFetcher':
return self
- def __exit__(self, exc_type, exc_value, traceback):
+ def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None:
+ assert self.conn is not None
+ self.conn.wait()
self.close()
""" Main indexing routine.
"""
- def __init__(self, dsn, tokenizer, num_threads):
+ def __init__(self, dsn: str, tokenizer: AbstractTokenizer, num_threads: int):
self.dsn = dsn
self.tokenizer = tokenizer
self.num_threads = num_threads
- def index_full(self, analyse=True):
- """ Index the complete database. This will first index boudnaries
+ def has_pending(self) -> bool:
+ """ Check if any data still needs indexing.
+ This function must only be used after the import has finished.
+ Otherwise it will be very expensive.
+ """
+ with connect(self.dsn) as conn:
+ with conn.cursor() as cur:
+ cur.execute("SELECT 'a' FROM placex WHERE indexed_status > 0 LIMIT 1")
+ return cur.rowcount > 0
+
+
+ def index_full(self, analyse: bool = True) -> None:
+ """ Index the complete database. This will first index boundaries
followed by all other objects. When `analyse` is True, then the
database will be analysed at the appropriate places to
ensure that database statistics are updated.
with connect(self.dsn) as conn:
conn.autocommit = True
- if analyse:
- def _analyze():
+ def _analyze() -> None:
+ if analyse:
with conn.cursor() as cur:
cur.execute('ANALYZE')
- else:
- def _analyze():
- pass
- self.index_by_rank(0, 4)
- _analyze()
+ if self.index_by_rank(0, 4) > 0:
+ _analyze()
- self.index_boundaries(0, 30)
- _analyze()
+ if self.index_boundaries(0, 30) > 100:
+ _analyze()
- self.index_by_rank(5, 25)
- _analyze()
+ if self.index_by_rank(5, 25) > 100:
+ _analyze()
- self.index_by_rank(26, 30)
- _analyze()
+ if self.index_by_rank(26, 30) > 1000:
+ _analyze()
- self.index_postcodes()
- _analyze()
+ if self.index_postcodes() > 100:
+ _analyze()
- def index_boundaries(self, minrank, maxrank):
+ def index_boundaries(self, minrank: int, maxrank: int) -> int:
""" Index only administrative boundaries within the given rank range.
"""
+ total = 0
LOG.warning("Starting indexing boundaries using %s threads",
self.num_threads)
with self.tokenizer.name_analyzer() as analyzer:
for rank in range(max(minrank, 4), min(maxrank, 26)):
- self._index(runners.BoundaryRunner(rank, analyzer))
+ total += self._index(runners.BoundaryRunner(rank, analyzer))
+
+ return total
- def index_by_rank(self, minrank, maxrank):
+ def index_by_rank(self, minrank: int, maxrank: int) -> int:
""" Index all entries of placex in the given rank range (inclusive)
in order of their address rank.
When rank 30 is requested then also interpolations and
places with address rank 0 will be indexed.
"""
+ total = 0
maxrank = min(maxrank, 30)
LOG.warning("Starting indexing rank (%i to %i) using %i threads",
minrank, maxrank, self.num_threads)
with self.tokenizer.name_analyzer() as analyzer:
- for rank in range(max(1, minrank), maxrank):
- self._index(runners.RankRunner(rank, analyzer))
+ for rank in range(max(1, minrank), maxrank + 1):
+ total += self._index(runners.RankRunner(rank, analyzer), 20 if rank == 30 else 1)
if maxrank == 30:
- self._index(runners.RankRunner(0, analyzer))
- self._index(runners.InterpolationRunner(analyzer), 20)
- self._index(runners.RankRunner(30, analyzer), 20)
- else:
- self._index(runners.RankRunner(maxrank, analyzer))
+ total += self._index(runners.RankRunner(0, analyzer))
+ total += self._index(runners.InterpolationRunner(analyzer), 20)
+ return total
- def index_postcodes(self):
- """Index the entries ofthe location_postcode table.
+
+ def index_postcodes(self) -> int:
+ """Index the entries of the location_postcode table.
"""
LOG.warning("Starting indexing postcodes using %s threads", self.num_threads)
- self._index(runners.PostcodeRunner(), 20)
+ return self._index(runners.PostcodeRunner(), 20)
- def update_status_table(self):
+ def update_status_table(self) -> None:
""" Update the status in the status table to 'indexed'.
"""
with connect(self.dsn) as conn:
conn.commit()
- def _index(self, runner, batch=1):
+ def _index(self, runner: runners.Runner, batch: int = 1) -> int:
""" 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
total_tuples = cur.scalar(runner.sql_count_objects())
LOG.debug("Total number of rows: %i", total_tuples)
- # need to fetch those manually because register_hstore cannot
- # fetch them on an asynchronous connection below.
- hstore_oid = cur.scalar("SELECT 'hstore'::regtype::oid")
- hstore_array_oid = cur.scalar("SELECT 'hstore[]'::regtype::oid")
-
conn.commit()
progress = ProgressLogger(runner.name(), total_tuples)
- fetcher_wait = 0
- pool_wait = 0
-
if total_tuples > 0:
with conn.cursor(name='places') as cur:
cur.execute(runner.sql_get_objects())
- fetcher = DBConnection(self.dsn, cursor_factory=psycopg2.extras.DictCursor)
- psycopg2.extras.register_hstore(fetcher.conn,
- oid=hstore_oid,
- array_oid=hstore_array_oid)
-
- with WorkerPool(self.dsn, self.num_threads) as pool:
- places = self._fetch_next_batch(cur, fetcher, runner)
- while places is not None:
- if not places:
- t0 = time.time()
- fetcher.wait()
- fetcher_wait += time.time() - t0
- places = fetcher.cursor.fetchall()
-
- # asynchronously get the next batch
- next_places = self._fetch_next_batch(cur, fetcher, runner)
-
- # And insert the curent batch
- for idx in range(0, len(places), batch):
- t0 = time.time()
- worker = pool.next_free_worker()
- pool_wait += time.time() - t0
- part = places[idx:idx+batch]
- LOG.debug("Processing places: %s", str(part))
- runner.index_places(worker, part)
- progress.add(len(part))
+ with PlaceFetcher(self.dsn, conn) as fetcher:
+ with WorkerPool(self.dsn, self.num_threads) as pool:
+ has_more = fetcher.fetch_next_batch(cur, runner)
+ while has_more:
+ places = fetcher.get_batch()
- places = next_places
+ # asynchronously get the next batch
+ has_more = fetcher.fetch_next_batch(cur, runner)
- pool.finish_all()
+ # And insert the current batch
+ for idx in range(0, len(places), batch):
+ part = places[idx:idx + batch]
+ LOG.debug("Processing places: %s", str(part))
+ runner.index_places(pool.next_free_worker(), part)
+ progress.add(len(part))
- fetcher.wait()
- fetcher.close()
+ LOG.info("Wait time: fetcher: %.2fs, pool: %.2fs",
+ fetcher.wait_time, pool.wait_time)
conn.commit()
- progress.done()
- LOG.warning("Wait time: fetcher: {}s, pool: {}s".format(fetcher_wait, pool_wait))
-
-
- def _fetch_next_batch(self, cur, fetcher, runner):
- ids = cur.fetchmany(100)
-
- if not ids:
- return None
-
- if not hasattr(runner, 'get_place_details'):
- return ids
-
- runner.get_place_details(fetcher, ids)
- return []
+ return progress.done()
projects / nominatim.git / blobdiff