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