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