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