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