nominatim/tools/database_import.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 Functions for setting up and importing a new Nominatim database.
   9 """
  10 from typing import Tuple, Optional, Union, Sequence, MutableMapping, Any
  11 import logging
  12 import os
  13 import selectors
  14 import subprocess
  15 from pathlib import Path
  16
  17 import psutil
  18 from psycopg2 import sql as pysql
  19
  20 from nominatim.config import Configuration
  21 from nominatim.db.connection import connect, get_pg_env, Connection
  22 from nominatim.db.async_connection import DBConnection
  23 from nominatim.db.sql_preprocessor import SQLPreprocessor
  24 from nominatim.tools.exec_utils import run_osm2pgsql
  25 from nominatim.errors import UsageError
  26 from nominatim.version import POSTGRESQL_REQUIRED_VERSION, POSTGIS_REQUIRED_VERSION, HSTORE_REQUIRED_VERSION
  27
  28 LOG = logging.getLogger()
  29
  30 def _require_version(module: str, actual: Tuple[int, int], expected: Tuple[int, int]) -> None:
  31     """ Compares the version for the given module and raises an exception
  32         if the actual version is too old.
  33     """
  34     if actual < expected:
  35         LOG.fatal('Minimum supported version of %s is %d.%d. '
  36                   'Found version %d.%d.',
  37                   module, expected[0], expected[1], actual[0], actual[1])
  38         raise UsageError(f'{module} is too old.')
  39
  40
  41 def check_existing_database_plugins(dsn: str):
  42     """ Check that the database has the required plugins installed."""
  43     with connect(dsn) as conn:
  44         _require_version('PostgreSQL server',
  45                          conn.server_version_tuple(),
  46                          POSTGRESQL_REQUIRED_VERSION)
  47         _require_version('PostGIS',
  48                          conn.postgis_version_tuple(),
  49                          POSTGIS_REQUIRED_VERSION)
  50         _require_version('hstore',
  51                          conn.hstore_version_tuple(),
  52                          HSTORE_REQUIRED_VERSION)
  53
  54
  55 def setup_database_skeleton(dsn: str, rouser: Optional[str] = None) -> None:
  56     """ Create a new database for Nominatim and populate it with the
  57         essential extensions.
  58
  59         The function fails when the database already exists or Postgresql or
  60         PostGIS versions are too old.
  61
  62         Uses `createdb` to create the database.
  63
  64         If 'rouser' is given, then the function also checks that the user
  65         with that given name exists.
  66
  67         Requires superuser rights by the caller.
  68     """
  69     proc = subprocess.run(['createdb'], env=get_pg_env(dsn), check=False)
  70
  71     if proc.returncode != 0:
  72         raise UsageError('Creating new database failed.')
  73
  74     with connect(dsn) as conn:
  75         _require_version('PostgreSQL server',
  76                          conn.server_version_tuple(),
  77                          POSTGRESQL_REQUIRED_VERSION)
  78
  79         if rouser is not None:
  80             with conn.cursor() as cur:
  81                 cnt = cur.scalar('SELECT count(*) FROM pg_user where usename = %s',
  82                                  (rouser, ))
  83                 if cnt == 0:
  84                     LOG.fatal("Web user '%s' does not exist. Create it with:\n"
  85                               "\n      createuser %s", rouser, rouser)
  86                     raise UsageError('Missing read-only user.')
  87
  88         # Create extensions.
  89         with conn.cursor() as cur:
  90             cur.execute('CREATE EXTENSION IF NOT EXISTS hstore')
  91             cur.execute('CREATE EXTENSION IF NOT EXISTS postgis')
  92
  93             postgis_version = conn.postgis_version_tuple()
  94             if postgis_version[0] >= 3:
  95                 cur.execute('CREATE EXTENSION IF NOT EXISTS postgis_raster')
  96
  97         conn.commit()
  98
  99         _require_version('PostGIS',
 100                          conn.postgis_version_tuple(),
 101                          POSTGIS_REQUIRED_VERSION)
 102
 103
 104 def import_osm_data(osm_files: Union[Path, Sequence[Path]],
 105                     options: MutableMapping[str, Any],
 106                     drop: bool = False, ignore_errors: bool = False) -> None:
 107     """ Import the given OSM files. 'options' contains the list of
 108         default settings for osm2pgsql.
 109     """
 110     options['import_file'] = osm_files
 111     options['append'] = False
 112     options['threads'] = 1
 113
 114     if not options['flatnode_file'] and options['osm2pgsql_cache'] == 0:
 115         # Make some educated guesses about cache size based on the size
 116         # of the import file and the available memory.
 117         mem = psutil.virtual_memory()
 118         fsize = 0
 119         if isinstance(osm_files, list):
 120             for fname in osm_files:
 121                 fsize += os.stat(str(fname)).st_size
 122         else:
 123             fsize = os.stat(str(osm_files)).st_size
 124         options['osm2pgsql_cache'] = int(min((mem.available + mem.cached) * 0.75,
 125                                              fsize * 2) / 1024 / 1024) + 1
 126
 127     run_osm2pgsql(options)
 128
 129     with connect(options['dsn']) as conn:
 130         if not ignore_errors:
 131             with conn.cursor() as cur:
 132                 cur.execute('SELECT * FROM place LIMIT 1')
 133                 if cur.rowcount == 0:
 134                     raise UsageError('No data imported by osm2pgsql.')
 135
 136         if drop:
 137             conn.drop_table('planet_osm_nodes')
 138
 139     if drop and options['flatnode_file']:
 140         Path(options['flatnode_file']).unlink()
 141
 142
 143 def create_tables(conn: Connection, config: Configuration, reverse_only: bool = False) -> None:
 144     """ Create the set of basic tables.
 145         When `reverse_only` is True, then the main table for searching will
 146         be skipped and only reverse search is possible.
 147     """
 148     sql = SQLPreprocessor(conn, config)
 149     sql.env.globals['db']['reverse_only'] = reverse_only
 150
 151     sql.run_sql_file(conn, 'tables.sql')
 152
 153
 154 def create_table_triggers(conn: Connection, config: Configuration) -> None:
 155     """ Create the triggers for the tables. The trigger functions must already
 156         have been imported with refresh.create_functions().
 157     """
 158     sql = SQLPreprocessor(conn, config)
 159     sql.run_sql_file(conn, 'table-triggers.sql')
 160
 161
 162 def create_partition_tables(conn: Connection, config: Configuration) -> None:
 163     """ Create tables that have explicit partitioning.
 164     """
 165     sql = SQLPreprocessor(conn, config)
 166     sql.run_sql_file(conn, 'partition-tables.src.sql')
 167
 168
 169 def truncate_data_tables(conn: Connection) -> None:
 170     """ Truncate all data tables to prepare for a fresh load.
 171     """
 172     with conn.cursor() as cur:
 173         cur.execute('TRUNCATE placex')
 174         cur.execute('TRUNCATE place_addressline')
 175         cur.execute('TRUNCATE location_area')
 176         cur.execute('TRUNCATE location_area_country')
 177         cur.execute('TRUNCATE location_property_tiger')
 178         cur.execute('TRUNCATE location_property_osmline')
 179         cur.execute('TRUNCATE location_postcode')
 180         if conn.table_exists('search_name'):
 181             cur.execute('TRUNCATE search_name')
 182         cur.execute('DROP SEQUENCE IF EXISTS seq_place')
 183         cur.execute('CREATE SEQUENCE seq_place start 100000')
 184
 185         cur.execute("""SELECT tablename FROM pg_tables
 186                        WHERE tablename LIKE 'location_road_%'""")
 187
 188         for table in [r[0] for r in list(cur)]:
 189             cur.execute('TRUNCATE ' + table)
 190
 191     conn.commit()
 192
 193
 194 _COPY_COLUMNS = pysql.SQL(',').join(map(pysql.Identifier,
 195                                         ('osm_type', 'osm_id', 'class', 'type',
 196                                          'name', 'admin_level', 'address',
 197                                          'extratags', 'geometry')))
 198
 199
 200 def load_data(dsn: str, threads: int) -> None:
 201     """ Copy data into the word and placex table.
 202     """
 203     sel = selectors.DefaultSelector()
 204     # Then copy data from place to placex in <threads - 1> chunks.
 205     place_threads = max(1, threads - 1)
 206     for imod in range(place_threads):
 207         conn = DBConnection(dsn)
 208         conn.connect()
 209         conn.perform(
 210             pysql.SQL("""INSERT INTO placex ({columns})
 211                            SELECT {columns} FROM place
 212                            WHERE osm_id % {total} = {mod}
 213                              AND NOT (class='place' and (type='houses' or type='postcode'))
 214                              AND ST_IsValid(geometry)
 215                       """).format(columns=_COPY_COLUMNS,
 216                                   total=pysql.Literal(place_threads),
 217                                   mod=pysql.Literal(imod)))
 218         sel.register(conn, selectors.EVENT_READ, conn)
 219
 220     # Address interpolations go into another table.
 221     conn = DBConnection(dsn)
 222     conn.connect()
 223     conn.perform("""INSERT INTO location_property_osmline (osm_id, address, linegeo)
 224                       SELECT osm_id, address, geometry FROM place
 225                       WHERE class='place' and type='houses' and osm_type='W'
 226                             and ST_GeometryType(geometry) = 'ST_LineString'
 227                  """)
 228     sel.register(conn, selectors.EVENT_READ, conn)
 229
 230     # Now wait for all of them to finish.
 231     todo = place_threads + 1
 232     while todo > 0:
 233         for key, _ in sel.select(1):
 234             conn = key.data
 235             sel.unregister(conn)
 236             conn.wait()
 237             conn.close()
 238             todo -= 1
 239         print('.', end='', flush=True)
 240     print('\n')
 241
 242     with connect(dsn) as syn_conn:
 243         with syn_conn.cursor() as cur:
 244             cur.execute('ANALYSE')
 245
 246
 247 def create_search_indices(conn: Connection, config: Configuration,
 248                           drop: bool = False, threads: int = 1) -> None:
 249     """ Create tables that have explicit partitioning.
 250     """
 251
 252     # If index creation failed and left an index invalid, they need to be
 253     # cleaned out first, so that the script recreates them.
 254     with conn.cursor() as cur:
 255         cur.execute("""SELECT relname FROM pg_class, pg_index
 256                        WHERE pg_index.indisvalid = false
 257                              AND pg_index.indexrelid = pg_class.oid""")
 258         bad_indices = [row[0] for row in list(cur)]
 259         for idx in bad_indices:
 260             LOG.info("Drop invalid index %s.", idx)
 261             cur.execute(pysql.SQL('DROP INDEX {}').format(pysql.Identifier(idx)))
 262     conn.commit()
 263
 264     sql = SQLPreprocessor(conn, config)
 265
 266     sql.run_parallel_sql_file(config.get_libpq_dsn(),
 267                               'indices.sql', min(8, threads), drop=drop)