test/python/indexer/test_indexing.py

   1 # SPDX-License-Identifier: GPL-3.0-or-later
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2025 by the Nominatim developer community.
   6 # For a full list of authors see the git log.
   7 """
   8 Tests for running the indexing.
   9 """
  10 import itertools
  11 import pytest
  12 import pytest_asyncio  # noqa
  13
  14 from nominatim_db.indexer import indexer
  15 from nominatim_db.tokenizer import factory
  16
  17
  18 class IndexerTestDB:
  19
  20     def __init__(self, conn):
  21         self.placex_id = itertools.count(100000)
  22         self.osmline_id = itertools.count(500000)
  23         self.postcode_id = itertools.count(700000)
  24
  25         self.conn = conn
  26         self.conn.autocimmit = True
  27         with self.conn.cursor() as cur:
  28             cur.execute("""CREATE TABLE placex (place_id BIGINT,
  29                                                 name HSTORE,
  30                                                 class TEXT,
  31                                                 type TEXT,
  32                                                 linked_place_id BIGINT,
  33                                                 rank_address SMALLINT,
  34                                                 rank_search SMALLINT,
  35                                                 indexed_status SMALLINT,
  36                                                 indexed_date TIMESTAMP,
  37                                                 partition SMALLINT,
  38                                                 admin_level SMALLINT,
  39                                                 country_code TEXT,
  40                                                 address HSTORE,
  41                                                 token_info JSONB,
  42                                                 geometry_sector INTEGER)""")
  43             cur.execute("""CREATE TABLE location_property_osmline (
  44                                place_id BIGINT,
  45                                osm_id BIGINT,
  46                                address HSTORE,
  47                                token_info JSONB,
  48                                indexed_status SMALLINT,
  49                                indexed_date TIMESTAMP,
  50                                geometry_sector INTEGER)""")
  51             cur.execute("""CREATE TABLE location_postcode (
  52                                place_id BIGINT,
  53                                indexed_status SMALLINT,
  54                                indexed_date TIMESTAMP,
  55                                country_code varchar(2),
  56                                postcode TEXT)""")
  57             cur.execute("""CREATE OR REPLACE FUNCTION date_update() RETURNS TRIGGER
  58                            AS $$
  59                            BEGIN
  60                              IF NEW.indexed_status = 0 and OLD.indexed_status != 0 THEN
  61                                NEW.indexed_date = now();
  62                              END IF;
  63                              RETURN NEW;
  64                            END; $$ LANGUAGE plpgsql;""")
  65             cur.execute("DROP TYPE IF EXISTS prepare_update_info CASCADE")
  66             cur.execute("""CREATE TYPE prepare_update_info AS (
  67                              name HSTORE,
  68                              address HSTORE,
  69                              rank_address SMALLINT,
  70                              country_code TEXT,
  71                              class TEXT,
  72                              type TEXT,
  73                              linked_place_id BIGINT
  74                            )""")
  75             cur.execute("""CREATE OR REPLACE FUNCTION placex_indexing_prepare(p placex,
  76                                                      OUT result prepare_update_info)
  77                            AS $$
  78                            BEGIN
  79                              result.address := p.address;
  80                              result.name := p.name;
  81                              result.class := p.class;
  82                              result.type := p.type;
  83                              result.country_code := p.country_code;
  84                              result.rank_address := p.rank_address;
  85                            END;
  86                            $$ LANGUAGE plpgsql STABLE;
  87                         """)
  88             cur.execute("""CREATE OR REPLACE FUNCTION
  89                              get_interpolation_address(in_address HSTORE, wayid BIGINT)
  90                            RETURNS HSTORE AS $$
  91                            BEGIN
  92                              RETURN in_address;
  93                            END;
  94                            $$ LANGUAGE plpgsql STABLE;
  95                         """)
  96
  97             for table in ('placex', 'location_property_osmline', 'location_postcode'):
  98                 cur.execute("""CREATE TRIGGER {0}_update BEFORE UPDATE ON {0}
  99                                FOR EACH ROW EXECUTE PROCEDURE date_update()
 100                             """.format(table))
 101
 102     def scalar(self, query):
 103         with self.conn.cursor() as cur:
 104             cur.execute(query)
 105             return cur.fetchone()[0]
 106
 107     def add_place(self, cls='place', typ='locality',
 108                   rank_search=30, rank_address=30, sector=20):
 109         next_id = next(self.placex_id)
 110         with self.conn.cursor() as cur:
 111             cur.execute("""INSERT INTO placex
 112                               (place_id, class, type, rank_search, rank_address,
 113                                indexed_status, geometry_sector)
 114                               VALUES (%s, %s, %s, %s, %s, 1, %s)""",
 115                         (next_id, cls, typ, rank_search, rank_address, sector))
 116         return next_id
 117
 118     def add_admin(self, **kwargs):
 119         kwargs['cls'] = 'boundary'
 120         kwargs['typ'] = 'administrative'
 121         return self.add_place(**kwargs)
 122
 123     def add_osmline(self, sector=20):
 124         next_id = next(self.osmline_id)
 125         with self.conn.cursor() as cur:
 126             cur.execute("""INSERT INTO location_property_osmline
 127                               (place_id, osm_id, indexed_status, geometry_sector)
 128                               VALUES (%s, %s, 1, %s)""",
 129                         (next_id, next_id, sector))
 130         return next_id
 131
 132     def add_postcode(self, country, postcode):
 133         next_id = next(self.postcode_id)
 134         with self.conn.cursor() as cur:
 135             cur.execute("""INSERT INTO location_postcode
 136                             (place_id, indexed_status, country_code, postcode)
 137                             VALUES (%s, 1, %s, %s)""",
 138                         (next_id, country, postcode))
 139         return next_id
 140
 141     def placex_unindexed(self):
 142         return self.scalar('SELECT count(*) from placex where indexed_status > 0')
 143
 144     def osmline_unindexed(self):
 145         return self.scalar("""SELECT count(*) from location_property_osmline
 146                               WHERE indexed_status > 0""")
 147
 148
 149 @pytest.fixture
 150 def test_db(temp_db_conn):
 151     yield IndexerTestDB(temp_db_conn)
 152
 153
 154 @pytest.fixture
 155 def test_tokenizer(tokenizer_mock, project_env):
 156     return factory.create_tokenizer(project_env)
 157
 158
 159 @pytest.mark.parametrize("threads", [1, 15])
 160 @pytest.mark.asyncio
 161 async def test_index_all_by_rank(test_db, threads, test_tokenizer):
 162     for rank in range(31):
 163         test_db.add_place(rank_address=rank, rank_search=rank)
 164     test_db.add_osmline()
 165
 166     assert test_db.placex_unindexed() == 31
 167     assert test_db.osmline_unindexed() == 1
 168
 169     idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, threads)
 170     await idx.index_by_rank(0, 30)
 171
 172     assert test_db.placex_unindexed() == 0
 173     assert test_db.osmline_unindexed() == 0
 174
 175     assert test_db.scalar("""SELECT count(*) from placex
 176                              WHERE indexed_status = 0 and indexed_date is null""") == 0
 177     # ranks come in order of rank address
 178     assert test_db.scalar("""
 179         SELECT count(*) FROM placex p WHERE rank_address > 0
 180           AND indexed_date >= (SELECT min(indexed_date) FROM placex o
 181                                WHERE p.rank_address < o.rank_address)""") == 0
 182     # placex address ranked objects come before interpolations
 183     assert test_db.scalar(
 184         """SELECT count(*) FROM placex WHERE rank_address > 0
 185              AND indexed_date >
 186                    (SELECT min(indexed_date) FROM location_property_osmline)""") == 0
 187     # rank 0 comes after all other placex objects
 188     assert test_db.scalar(
 189         """SELECT count(*) FROM placex WHERE rank_address > 0
 190              AND indexed_date >
 191                    (SELECT min(indexed_date) FROM placex WHERE rank_address = 0)""") == 0
 192
 193
 194 @pytest.mark.parametrize("threads", [1, 15])
 195 @pytest.mark.asyncio
 196 async def test_index_partial_without_30(test_db, threads, test_tokenizer):
 197     for rank in range(31):
 198         test_db.add_place(rank_address=rank, rank_search=rank)
 199     test_db.add_osmline()
 200
 201     assert test_db.placex_unindexed() == 31
 202     assert test_db.osmline_unindexed() == 1
 203
 204     idx = indexer.Indexer('dbname=test_nominatim_python_unittest',
 205                           test_tokenizer, threads)
 206     await idx.index_by_rank(4, 15)
 207
 208     assert test_db.placex_unindexed() == 19
 209     assert test_db.osmline_unindexed() == 1
 210
 211     assert test_db.scalar("""
 212                     SELECT count(*) FROM placex
 213                       WHERE indexed_status = 0 AND not rank_address between 4 and 15""") == 0
 214
 215
 216 @pytest.mark.parametrize("threads", [1, 15])
 217 @pytest.mark.asyncio
 218 async def test_index_partial_with_30(test_db, threads, test_tokenizer):
 219     for rank in range(31):
 220         test_db.add_place(rank_address=rank, rank_search=rank)
 221     test_db.add_osmline()
 222
 223     assert test_db.placex_unindexed() == 31
 224     assert test_db.osmline_unindexed() == 1
 225
 226     idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, threads)
 227     await idx.index_by_rank(28, 30)
 228
 229     assert test_db.placex_unindexed() == 27
 230     assert test_db.osmline_unindexed() == 0
 231
 232     assert test_db.scalar("""
 233                     SELECT count(*) FROM placex
 234                       WHERE indexed_status = 0 AND rank_address between 1 and 27""") == 0
 235
 236
 237 @pytest.mark.parametrize("threads", [1, 15])
 238 @pytest.mark.asyncio
 239 async def test_index_boundaries(test_db, threads, test_tokenizer):
 240     for rank in range(4, 10):
 241         test_db.add_admin(rank_address=rank, rank_search=rank)
 242     for rank in range(31):
 243         test_db.add_place(rank_address=rank, rank_search=rank)
 244     test_db.add_osmline()
 245
 246     assert test_db.placex_unindexed() == 37
 247     assert test_db.osmline_unindexed() == 1
 248
 249     idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, threads)
 250     await idx.index_boundaries(0, 30)
 251
 252     assert test_db.placex_unindexed() == 31
 253     assert test_db.osmline_unindexed() == 1
 254
 255     assert test_db.scalar("""
 256                     SELECT count(*) FROM placex
 257                       WHERE indexed_status = 0 AND class != 'boundary'""") == 0
 258
 259
 260 @pytest.mark.parametrize("threads", [1, 15])
 261 @pytest.mark.asyncio
 262 async def test_index_postcodes(test_db, threads, test_tokenizer):
 263     for postcode in range(1000):
 264         test_db.add_postcode('de', postcode)
 265     for postcode in range(32000, 33000):
 266         test_db.add_postcode('us', postcode)
 267
 268     idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, threads)
 269     await idx.index_postcodes()
 270
 271     assert test_db.scalar("""SELECT count(*) FROM location_postcode
 272                                   WHERE indexed_status != 0""") == 0
 273
 274
 275 @pytest.mark.parametrize("analyse", [True, False])
 276 @pytest.mark.asyncio
 277 async def test_index_full(test_db, analyse, test_tokenizer):
 278     for rank in range(4, 10):
 279         test_db.add_admin(rank_address=rank, rank_search=rank)
 280     for rank in range(31):
 281         test_db.add_place(rank_address=rank, rank_search=rank)
 282     test_db.add_osmline()
 283     for postcode in range(1000):
 284         test_db.add_postcode('de', postcode)
 285
 286     idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, 4)
 287     await idx.index_full(analyse=analyse)
 288
 289     assert test_db.placex_unindexed() == 0
 290     assert test_db.osmline_unindexed() == 0
 291     assert test_db.scalar("""SELECT count(*) FROM location_postcode
 292                              WHERE indexed_status != 0""") == 0