From c5ff2143c48ebd46b94c353fb83ff2e1212c7a9e Mon Sep 17 00:00:00 2001 From: Sarah Hoffmann Date: Thu, 28 May 2026 14:47:31 +0200 Subject: [PATCH] refactor index runner to remove analyzer dependency --- src/nominatim_db/indexer/indexer.py | 100 +++++++++++++++------------- src/nominatim_db/indexer/runners.py | 35 ++-------- 2 files changed, 58 insertions(+), 77 deletions(-) diff --git a/src/nominatim_db/indexer/indexer.py b/src/nominatim_db/indexer/indexer.py index 4e7c831b..bf1ba2c1 100644 --- a/src/nominatim_db/indexer/indexer.py +++ b/src/nominatim_db/indexer/indexer.py @@ -2,17 +2,19 @@ # # This file is part of Nominatim. (https://nominatim.org) # -# Copyright (C) 2025 by the Nominatim developer community. +# Copyright (C) 2026 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 cast, List, Any, Optional +from typing import cast, Any, Optional import logging import time import psycopg +from psycopg.types.json import Json +from ..data.place_info import PlaceInfo from ..db.connection import connect, execute_scalar from ..db.query_pool import QueryPool from ..tokenizer.base import AbstractTokenizer @@ -109,10 +111,9 @@ class Indexer: (minrank, maxrank)) total_tuples = {row.rank_search: row.count for row in cur} - with self.tokenizer.name_analyzer() as analyzer: - for rank in range(minrank, maxrank + 1): - total += await self._index(runners.BoundaryRunner(rank, analyzer), - total_tuples=total_tuples.get(rank, 0)) + for rank in range(minrank, maxrank + 1): + total += await self._index(runners.BoundaryRunner(rank), + total_tuples=total_tuples.get(rank, 0)) return total @@ -144,23 +145,22 @@ class Indexer: (minrank, maxrank)) total_tuples = {row.rank_address: row.count for row in cur} - with self.tokenizer.name_analyzer() as analyzer: - for rank in range(max(1, minrank), maxrank + 1): - if rank >= 30: - batch = 20 - elif rank >= 26: - batch = 5 - else: - batch = 1 - total += await self._index(runners.RankRunner(rank, analyzer), - batch=batch, total_tuples=total_tuples.get(rank, 0)) + for rank in range(max(1, minrank), maxrank + 1): + if rank >= 30: + batch = 20 + elif rank >= 26: + batch = 5 + else: + batch = 1 + total += await self._index(runners.RankRunner(rank), + batch=batch, total_tuples=total_tuples.get(rank, 0)) - # Special case: rank zero depends on ranks [1..30] - if minrank == 0: - total += await self._index(runners.RankRunner(0, analyzer)) + # Special case: rank zero depends on ranks [1..30] + if minrank == 0: + total += await self._index(runners.RankRunner(0)) - if maxrank == 30: - total += await self._index(runners.InterpolationRunner(analyzer), batch=20) + if maxrank == 30: + total += await self._index(runners.InterpolationRunner(), batch=20) return total @@ -198,32 +198,38 @@ class Indexer: progress = ProgressLogger(runner.name(), total_tuples) if total_tuples > 0: - async with await psycopg.AsyncConnection.connect( - self.dsn, row_factory=psycopg.rows.dict_row) as aconn, \ - QueryPool(self.dsn, self.num_threads, autocommit=True) as pool: - fetcher_time = 0.0 - tstart = time.time() - async with aconn.cursor(name='places') as cur: - query = runner.index_places_query(batch) - params: List[Any] = [] - num_places = 0 - async for place in cur.stream(runner.sql_get_objects()): - fetcher_time += time.time() - tstart - - params.extend(runner.index_places_params(place)) - num_places += 1 - - if num_places >= batch: - LOG.debug("Processing places: %s", str(params)) - await pool.put_query(query, params) - progress.add(num_places) - params = [] - num_places = 0 - - tstart = time.time() - - if num_places > 0: - await pool.put_query(runner.index_places_query(num_places), params) + with self.tokenizer.name_analyzer() as analyzer: + async with await psycopg.AsyncConnection.connect( + self.dsn, row_factory=psycopg.rows.dict_row) as aconn, \ + QueryPool(self.dsn, self.num_threads, autocommit=True) as pool: + fetcher_time = 0.0 + tstart = time.time() + async with aconn.cursor(name='places') as cur: + query = runner.index_places_query(batch) + params: list[Any] = [] + num_places = 0 + needs_token_info = 'token_info' in runner.QUERY_ROWS + async for place in cur.stream(runner.sql_get_objects()): + fetcher_time += time.time() - tstart + + if needs_token_info: + place_info = PlaceInfo(place) + place['token_info'] = Json(analyzer.process_place(place_info)) + + params.extend(place.get(i) for i in runner.QUERY_ROWS) + num_places += 1 + + if num_places >= batch: + LOG.debug("Processing places: %s", str(params)) + await pool.put_query(query, params) + progress.add(num_places) + params = [] + num_places = 0 + + tstart = time.time() + + if num_places > 0: + await pool.put_query(runner.index_places_query(num_places), params) LOG.info("Wait time: fetcher: %.2fs, pool: %.2fs", fetcher_time, pool.wait_time) diff --git a/src/nominatim_db/indexer/runners.py b/src/nominatim_db/indexer/runners.py index e3e0e0f4..8e80db02 100644 --- a/src/nominatim_db/indexer/runners.py +++ b/src/nominatim_db/indexer/runners.py @@ -8,31 +8,21 @@ Mix-ins that provide the actual commands for the indexer for various indexing tasks. """ -from typing import Any, Sequence - from psycopg import sql as pysql -from psycopg.rows import DictRow -from psycopg.types.json import Json from ..typing import Protocol, QueryNoTemplate -from ..data.place_info import PlaceInfo -from ..tokenizer.base import AbstractAnalyzer def _mk_valuelist(template: str, num: int) -> pysql.Composed: return pysql.SQL(',').join([pysql.SQL(template)] * num) -def _analyze_place(place: DictRow, analyzer: AbstractAnalyzer) -> Json: - return Json(analyzer.process_place(PlaceInfo(place))) - - class Runner(Protocol): + QUERY_ROWS: list[str] = [] def name(self) -> str: ... def sql_count_objects(self) -> QueryNoTemplate: ... def sql_get_objects(self) -> QueryNoTemplate: ... def index_places_query(self, batch_size: int) -> QueryNoTemplate: ... - def index_places_params(self, place: DictRow) -> Sequence[Any]: ... SELECT_SQL = pysql.SQL("""SELECT place_id, extra.* @@ -44,10 +34,10 @@ UPDATE_LINE = "(%s, %s::hstore, %s::hstore, %s::int, %s::jsonb)" class AbstractPlacexRunner: """ Returns SQL commands for indexing of the placex table. """ + QUERY_ROWS = ['place_id', 'name', 'address', 'linked_place_id', 'token_info'] - def __init__(self, rank: int, analyzer: AbstractAnalyzer) -> None: + def __init__(self, rank: int) -> None: self.rank = rank - self.analyzer = analyzer def index_places_query(self, batch_size: int) -> QueryNoTemplate: return pysql.SQL( @@ -58,13 +48,6 @@ class AbstractPlacexRunner: WHERE place_id = v.id """).format(_mk_valuelist(UPDATE_LINE, batch_size)) - def index_places_params(self, place: DictRow) -> Sequence[Any]: - return (place['place_id'], - place['name'], - place['address'], - place['linked_place_id'], - _analyze_place(place, self.analyzer)) - class RankRunner(AbstractPlacexRunner): """ Returns SQL commands for indexing one rank within the placex table. @@ -112,9 +95,7 @@ class InterpolationRunner: """ Returns SQL commands for indexing the address interpolation table location_property_osmline. """ - - def __init__(self, analyzer: AbstractAnalyzer) -> None: - self.analyzer = analyzer + QUERY_ROWS = ['place_id', 'address', 'token_info'] def name(self) -> str: return "interpolation lines (location_property_osmline)" @@ -136,14 +117,11 @@ class InterpolationRunner: WHERE place_id = v.id """).format(_mk_valuelist("(%s, %s::hstore, %s::jsonb)", batch_size)) - def index_places_params(self, place: DictRow) -> Sequence[Any]: - return (place['place_id'], place['address'], - _analyze_place(place, self.analyzer)) - class PostcodeRunner(Runner): """ Provides the SQL commands for indexing the location_postcodes table. """ + QUERY_ROWS = ['place_id'] def name(self) -> str: return "postcodes (location_postcodes)" @@ -160,6 +138,3 @@ class PostcodeRunner(Runner): return pysql.SQL("""UPDATE location_postcodes SET indexed_status = 0 WHERE place_id IN ({})""")\ .format(pysql.SQL(',').join((pysql.Placeholder() for _ in range(batch_size)))) - - def index_places_params(self, place: DictRow) -> Sequence[Any]: - return (place['place_id'], ) -- 2.47.3