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