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 import pytest_asyncio
  13
  14 from nominatim_db.indexer import indexer
  15 from nominatim_db.tokenizer import factory
  16
  17 class IndexerTestDB:
  18
  19     def __init__(self, conn):
  20         self.placex_id = itertools.count(100000)
  21         self.osmline_id = itertools.count(500000)
  22         self.postcode_id = itertools.count(700000)
  23
  24         self.conn = conn
  25         self.conn.autocimmit = True
  26         with self.conn.cursor() as cur:
  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 @pytest.mark.asyncio
 160 async def test_index_all_by_rank(test_db, threads, test_tokenizer):
 161     for rank in range(31):
 162         test_db.add_place(rank_address=rank, rank_search=rank)
 163     test_db.add_osmline()
 164
 165     assert test_db.placex_unindexed() == 31
 166     assert test_db.osmline_unindexed() == 1
 167
 168     idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, threads)
 169     await idx.index_by_rank(0, 30)
 170
 171     assert test_db.placex_unindexed() == 0
 172     assert test_db.osmline_unindexed() == 0
 173
 174     assert test_db.scalar("""SELECT count(*) from placex
 175                              WHERE indexed_status = 0 and indexed_date is null""") == 0
 176     # ranks come in order of rank address
 177     assert test_db.scalar("""
 178         SELECT count(*) FROM placex p WHERE rank_address > 0
 179           AND indexed_date >= (SELECT min(indexed_date) FROM placex o
 180                                WHERE p.rank_address < o.rank_address)""") == 0
 181     # placex address ranked objects come before interpolations
 182     assert test_db.scalar(
 183         """SELECT count(*) FROM placex WHERE rank_address > 0
 184              AND indexed_date >
 185                    (SELECT min(indexed_date) FROM location_property_osmline)""") == 0
 186     # rank 0 comes after all other placex objects
 187     assert test_db.scalar(
 188         """SELECT count(*) FROM placex WHERE rank_address > 0
 189              AND indexed_date >
 190                    (SELECT min(indexed_date) FROM placex WHERE rank_address = 0)""") == 0
 191
 192
 193 @pytest.mark.parametrize("threads", [1, 15])
 194 @pytest.mark.asyncio
 195 async def test_index_partial_without_30(test_db, threads, test_tokenizer):
 196     for rank in range(31):
 197         test_db.add_place(rank_address=rank, rank_search=rank)
 198     test_db.add_osmline()
 199
 200     assert test_db.placex_unindexed() == 31
 201     assert test_db.osmline_unindexed() == 1
 202
 203     idx = indexer.Indexer('dbname=test_nominatim_python_unittest',
 204                           test_tokenizer, threads)
 205     await idx.index_by_rank(4, 15)
 206
 207     assert test_db.placex_unindexed() == 19
 208     assert test_db.osmline_unindexed() == 1
 209
 210     assert test_db.scalar("""
 211                     SELECT count(*) FROM placex
 212                       WHERE indexed_status = 0 AND not rank_address between 4 and 15""") == 0
 213
 214
 215 @pytest.mark.parametrize("threads", [1, 15])
 216 @pytest.mark.asyncio
 217 async def test_index_partial_with_30(test_db, threads, test_tokenizer):
 218     for rank in range(31):
 219         test_db.add_place(rank_address=rank, rank_search=rank)
 220     test_db.add_osmline()
 221
 222     assert test_db.placex_unindexed() == 31
 223     assert test_db.osmline_unindexed() == 1
 224
 225     idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, threads)
 226     await idx.index_by_rank(28, 30)
 227
 228     assert test_db.placex_unindexed() == 27
 229     assert test_db.osmline_unindexed() == 0
 230
 231     assert test_db.scalar("""
 232                     SELECT count(*) FROM placex
 233                       WHERE indexed_status = 0 AND rank_address between 1 and 27""") == 0
 234
 235 @pytest.mark.parametrize("threads", [1, 15])
 236 @pytest.mark.asyncio
 237 async def test_index_boundaries(test_db, threads, test_tokenizer):
 238     for rank in range(4, 10):
 239         test_db.add_admin(rank_address=rank, rank_search=rank)
 240     for rank in range(31):
 241         test_db.add_place(rank_address=rank, rank_search=rank)
 242     test_db.add_osmline()
 243
 244     assert test_db.placex_unindexed() == 37
 245     assert test_db.osmline_unindexed() == 1
 246
 247     idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, threads)
 248     await idx.index_boundaries(0, 30)
 249
 250     assert test_db.placex_unindexed() == 31
 251     assert test_db.osmline_unindexed() == 1
 252
 253     assert test_db.scalar("""
 254                     SELECT count(*) FROM placex
 255                       WHERE indexed_status = 0 AND class != 'boundary'""") == 0
 256
 257
 258 @pytest.mark.parametrize("threads", [1, 15])
 259 @pytest.mark.asyncio
 260 async def test_index_postcodes(test_db, threads, test_tokenizer):
 261     for postcode in range(1000):
 262         test_db.add_postcode('de', postcode)
 263     for postcode in range(32000, 33000):
 264         test_db.add_postcode('us', postcode)
 265
 266     idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, threads)
 267     await idx.index_postcodes()
 268
 269     assert test_db.scalar("""SELECT count(*) FROM location_postcode
 270                                   WHERE indexed_status != 0""") == 0
 271
 272
 273 @pytest.mark.parametrize("analyse", [True, False])
 274 @pytest.mark.asyncio
 275 async def test_index_full(test_db, analyse, test_tokenizer):
 276     for rank in range(4, 10):
 277         test_db.add_admin(rank_address=rank, rank_search=rank)
 278     for rank in range(31):
 279         test_db.add_place(rank_address=rank, rank_search=rank)
 280     test_db.add_osmline()
 281     for postcode in range(1000):
 282         test_db.add_postcode('de', postcode)
 283
 284     idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, 4)
 285     await idx.index_full(analyse=analyse)
 286
 287     assert test_db.placex_unindexed() == 0
 288     assert test_db.osmline_unindexed() == 0
 289     assert test_db.scalar("""SELECT count(*) FROM location_postcode
 290                              WHERE indexed_status != 0""") == 0