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