test/python/indexer/test_indexing.py

   1 #
   2 # This file is part of Nominatim. (https://nominatim.org)
   3 #
   4 # Copyright (C) 2026 by the Nominatim developer community.
   5 # For a full list of authors see the git log.
   6 """
   7 Tests for running the indexing.
   8 """
   9
  10 import pytest
  11 import pytest_asyncio  # noqa
  12
  13 from nominatim_db.indexer import indexer
  14 from nominatim_db.tokenizer import factory
  15
  16
  17 class TestIndexing:
  18     @pytest.fixture(autouse=True)
  19     def setup(self, temp_db_conn, project_env, tokenizer_mock,
  20               placex_table, postcode_table, osmline_table):
  21         self.conn = temp_db_conn
  22         temp_db_conn.execute("""
  23             CREATE OR REPLACE FUNCTION date_update() RETURNS TRIGGER AS $$
  24             BEGIN
  25               IF NEW.indexed_status = 0 and OLD.indexed_status != 0 THEN
  26                 NEW.indexed_date = now();
  27               END IF;
  28               RETURN NEW;
  29             END; $$ LANGUAGE plpgsql;
  30
  31             DROP TYPE IF EXISTS prepare_update_info CASCADE;
  32             CREATE TYPE prepare_update_info AS (
  33                          name HSTORE,
  34                          address HSTORE,
  35                          rank_address SMALLINT,
  36                          country_code TEXT,
  37                          class TEXT,
  38                          type TEXT,
  39                          linked_place_id BIGINT
  40                        );
  41             CREATE OR REPLACE FUNCTION placex_indexing_prepare(p placex,
  42                                                  OUT result prepare_update_info) AS $$
  43             BEGIN
  44               result.address := p.address;
  45               result.name := p.name;
  46               result.class := p.class;
  47               result.type := p.type;
  48               result.country_code := p.country_code;
  49               result.rank_address := p.rank_address;
  50             END; $$ LANGUAGE plpgsql STABLE;
  51
  52             CREATE OR REPLACE FUNCTION get_interpolation_address(in_address HSTORE, wayid BIGINT)
  53             RETURNS HSTORE AS $$ SELECT in_address $$ LANGUAGE sql STABLE;
  54         """)
  55
  56         for table in ('placex', 'location_property_osmline', 'location_postcodes'):
  57             temp_db_conn.execute("""CREATE TRIGGER {0}_update BEFORE UPDATE ON {0}
  58                                     FOR EACH ROW EXECUTE PROCEDURE date_update()
  59                                  """.format(table))
  60
  61         self.tokenizer = factory.create_tokenizer(project_env)
  62
  63     def scalar(self, query):
  64         with self.conn.cursor() as cur:
  65             cur.execute(query)
  66             return cur.fetchone()[0]
  67
  68     def placex_unindexed(self):
  69         return self.scalar('SELECT count(*) from placex where indexed_status > 0')
  70
  71     def osmline_unindexed(self):
  72         return self.scalar("""SELECT count(*) from location_property_osmline
  73                               WHERE indexed_status > 0""")
  74
  75     @pytest.mark.parametrize("threads", [1, 15])
  76     @pytest.mark.asyncio
  77     async def test_index_all_by_rank(self, dsn, threads, placex_row, osmline_row):
  78         for rank in range(31):
  79             placex_row(rank_address=rank, rank_search=rank, indexed_status=1)
  80         osmline_row()
  81
  82         assert self.placex_unindexed() == 31
  83         assert self.osmline_unindexed() == 1
  84
  85         idx = indexer.Indexer(dsn, self.tokenizer, threads)
  86         await idx.index_by_rank(0, 30)
  87
  88         assert self.placex_unindexed() == 0
  89         assert self.osmline_unindexed() == 0
  90
  91         assert self.scalar("""SELECT count(*) from placex
  92                                  WHERE indexed_status = 0 and indexed_date is null""") == 0
  93         # ranks come in order of rank address
  94         assert self.scalar("""
  95             SELECT count(*) FROM placex p WHERE rank_address > 0
  96               AND indexed_date >= (SELECT min(indexed_date) FROM placex o
  97                                    WHERE p.rank_address < o.rank_address)""") == 0
  98         # placex address ranked objects come before interpolations
  99         assert self.scalar(
 100             """SELECT count(*) FROM placex WHERE rank_address > 0
 101                  AND indexed_date >
 102                        (SELECT min(indexed_date) FROM location_property_osmline)""") == 0
 103         # rank 0 comes after all other placex objects
 104         assert self.scalar(
 105             """SELECT count(*) FROM placex WHERE rank_address > 0
 106                  AND indexed_date >
 107                        (SELECT min(indexed_date) FROM placex WHERE rank_address = 0)""") == 0
 108
 109     @pytest.mark.parametrize("threads", [1, 15])
 110     @pytest.mark.asyncio
 111     async def test_index_partial_without_30(self, dsn, threads, placex_row, osmline_row):
 112         for rank in range(31):
 113             placex_row(rank_address=rank, rank_search=rank, indexed_status=1)
 114         osmline_row()
 115
 116         assert self.placex_unindexed() == 31
 117         assert self.osmline_unindexed() == 1
 118
 119         idx = indexer.Indexer(dsn, self.tokenizer, threads)
 120         await idx.index_by_rank(4, 15)
 121
 122         assert self.placex_unindexed() == 19
 123         assert self.osmline_unindexed() == 1
 124
 125         assert self.scalar("""
 126                         SELECT count(*) FROM placex
 127                           WHERE indexed_status = 0 AND not rank_address between 4 and 15""") == 0
 128
 129     @pytest.mark.parametrize("threads", [1, 15])
 130     @pytest.mark.asyncio
 131     async def test_index_partial_with_30(self, dsn, threads, placex_row, osmline_row):
 132         for rank in range(31):
 133             placex_row(rank_address=rank, rank_search=rank, indexed_status=1)
 134         osmline_row()
 135
 136         assert self.placex_unindexed() == 31
 137         assert self.osmline_unindexed() == 1
 138
 139         idx = indexer.Indexer(dsn, self.tokenizer, threads)
 140         await idx.index_by_rank(28, 30)
 141
 142         assert self.placex_unindexed() == 28
 143         assert self.osmline_unindexed() == 0
 144
 145         assert self.scalar("""
 146                         SELECT count(*) FROM placex
 147                           WHERE indexed_status = 0 AND rank_address between 0 and 27""") == 0
 148
 149     @pytest.mark.parametrize("threads", [1, 15])
 150     @pytest.mark.asyncio
 151     async def test_index_boundaries(self, dsn, threads, placex_row, osmline_row):
 152         for rank in range(4, 10):
 153             placex_row(cls='boundary', typ='administrative',
 154                        rank_address=rank, rank_search=rank, indexed_status=1)
 155         for rank in range(31):
 156             placex_row(rank_address=rank, rank_search=rank, indexed_status=1)
 157         osmline_row()
 158
 159         assert self.placex_unindexed() == 37
 160         assert self.osmline_unindexed() == 1
 161
 162         idx = indexer.Indexer(dsn, self.tokenizer, threads)
 163         await idx.index_boundaries()
 164
 165         assert self.placex_unindexed() == 31
 166         assert self.osmline_unindexed() == 1
 167
 168         assert self.scalar("""
 169                         SELECT count(*) FROM placex
 170                           WHERE indexed_status = 0 AND class != 'boundary'""") == 0
 171
 172     @pytest.mark.parametrize("threads", [1, 15])
 173     @pytest.mark.asyncio
 174     async def test_index_postcodes(self, dsn, threads, postcode_row):
 175         for postcode in range(1000):
 176             postcode_row(country='de', postcode=postcode)
 177         for postcode in range(32000, 33000):
 178             postcode_row(country='us', postcode=postcode)
 179
 180         idx = indexer.Indexer(dsn, self.tokenizer, threads)
 181         await idx.index_postcodes()
 182
 183         assert self.scalar("""SELECT count(*) FROM location_postcodes
 184                                       WHERE indexed_status != 0""") == 0
 185
 186     @pytest.mark.parametrize("analyse", [True, False])
 187     @pytest.mark.asyncio
 188     async def test_index_full(self, dsn, analyse, placex_row, osmline_row, postcode_row):
 189         for rank in range(4, 10):
 190             placex_row(cls='boundary', typ='administrative',
 191                        rank_address=rank, rank_search=rank, indexed_status=1)
 192         for rank in range(31):
 193             placex_row(rank_address=rank, rank_search=rank, indexed_status=1)
 194         osmline_row()
 195         for postcode in range(1000):
 196             postcode_row(country='de', postcode=postcode)
 197
 198         idx = indexer.Indexer(dsn, self.tokenizer, 4)
 199         await idx.index_full(analyse=analyse)
 200
 201         assert self.placex_unindexed() == 0
 202         assert self.osmline_unindexed() == 0
 203         assert self.scalar("""SELECT count(*) FROM location_postcodes
 204                                  WHERE indexed_status != 0""") == 0