test/python/indexer/test_indexing.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # This file is part of Nominatim. (https://nominatim.org)
   4 #
   5 # Copyright (C) 2022 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.indexer import indexer
  14 from nominatim.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 rank < 30 objects come before interpolations
 181     assert test_db.scalar(
 182         """SELECT count(*) FROM placex WHERE rank_address < 30
 183              AND indexed_date >
 184                    (SELECT min(indexed_date) FROM location_property_osmline)""") == 0
 185     # placex rank = 30 objects come after interpolations
 186     assert test_db.scalar(
 187         """SELECT count(*) FROM placex WHERE rank_address = 30
 188              AND indexed_date <
 189                    (SELECT max(indexed_date) FROM location_property_osmline)""") == 0
 190     # rank 0 comes after rank 29 and before rank 30
 191     assert test_db.scalar(
 192         """SELECT count(*) FROM placex WHERE rank_address < 30
 193              AND indexed_date >
 194                    (SELECT min(indexed_date) FROM placex WHERE rank_address = 0)""") == 0
 195     assert test_db.scalar(
 196         """SELECT count(*) FROM placex WHERE rank_address = 30
 197              AND indexed_date <
 198                    (SELECT max(indexed_date) FROM placex WHERE rank_address = 0)""") == 0
 199
 200
 201 @pytest.mark.parametrize("threads", [1, 15])
 202 def test_index_partial_without_30(test_db, threads, test_tokenizer):
 203     for rank in range(31):
 204         test_db.add_place(rank_address=rank, rank_search=rank)
 205     test_db.add_osmline()
 206
 207     assert test_db.placex_unindexed() == 31
 208     assert test_db.osmline_unindexed() == 1
 209
 210     idx = indexer.Indexer('dbname=test_nominatim_python_unittest',
 211                           test_tokenizer, threads)
 212     idx.index_by_rank(4, 15)
 213
 214     assert test_db.placex_unindexed() == 19
 215     assert test_db.osmline_unindexed() == 1
 216
 217     assert test_db.scalar("""
 218                     SELECT count(*) FROM placex
 219                       WHERE indexed_status = 0 AND not rank_address between 4 and 15""") == 0
 220
 221
 222 @pytest.mark.parametrize("threads", [1, 15])
 223 def test_index_partial_with_30(test_db, threads, test_tokenizer):
 224     for rank in range(31):
 225         test_db.add_place(rank_address=rank, rank_search=rank)
 226     test_db.add_osmline()
 227
 228     assert test_db.placex_unindexed() == 31
 229     assert test_db.osmline_unindexed() == 1
 230
 231     idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, threads)
 232     idx.index_by_rank(28, 30)
 233
 234     assert test_db.placex_unindexed() == 27
 235     assert test_db.osmline_unindexed() == 0
 236
 237     assert test_db.scalar("""
 238                     SELECT count(*) FROM placex
 239                       WHERE indexed_status = 0 AND rank_address between 1 and 27""") == 0
 240
 241 @pytest.mark.parametrize("threads", [1, 15])
 242 def test_index_boundaries(test_db, threads, test_tokenizer):
 243     for rank in range(4, 10):
 244         test_db.add_admin(rank_address=rank, rank_search=rank)
 245     for rank in range(31):
 246         test_db.add_place(rank_address=rank, rank_search=rank)
 247     test_db.add_osmline()
 248
 249     assert test_db.placex_unindexed() == 37
 250     assert test_db.osmline_unindexed() == 1
 251
 252     idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, threads)
 253     idx.index_boundaries(0, 30)
 254
 255     assert test_db.placex_unindexed() == 31
 256     assert test_db.osmline_unindexed() == 1
 257
 258     assert test_db.scalar("""
 259                     SELECT count(*) FROM placex
 260                       WHERE indexed_status = 0 AND class != 'boundary'""") == 0
 261
 262
 263 @pytest.mark.parametrize("threads", [1, 15])
 264 def test_index_postcodes(test_db, threads, test_tokenizer):
 265     for postcode in range(1000):
 266         test_db.add_postcode('de', postcode)
 267     for postcode in range(32000, 33000):
 268         test_db.add_postcode('us', postcode)
 269
 270     idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, threads)
 271     idx.index_postcodes()
 272
 273     assert test_db.scalar("""SELECT count(*) FROM location_postcode
 274                                   WHERE indexed_status != 0""") == 0
 275
 276
 277 @pytest.mark.parametrize("analyse", [True, False])
 278 def test_index_full(test_db, analyse, test_tokenizer):
 279     for rank in range(4, 10):
 280         test_db.add_admin(rank_address=rank, rank_search=rank)
 281     for rank in range(31):
 282         test_db.add_place(rank_address=rank, rank_search=rank)
 283     test_db.add_osmline()
 284     for postcode in range(1000):
 285         test_db.add_postcode('de', postcode)
 286
 287     idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, 4)
 288     idx.index_full(analyse=analyse)
 289
 290     assert test_db.placex_unindexed() == 0
 291     assert test_db.osmline_unindexed() == 0
 292     assert test_db.scalar("""SELECT count(*) FROM location_postcode
 293                              WHERE indexed_status != 0""") == 0
 294
 295
 296 @pytest.mark.parametrize("threads", [1, 15])
 297 def test_index_reopen_connection(test_db, threads, monkeypatch, test_tokenizer):
 298     monkeypatch.setattr(indexer.WorkerPool, "REOPEN_CONNECTIONS_AFTER", 15)
 299
 300     for _ in range(1000):
 301         test_db.add_place(rank_address=30, rank_search=30)
 302
 303     idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, threads)
 304     idx.index_by_rank(28, 30)
 305
 306     assert test_db.placex_unindexed() == 0