]> git.openstreetmap.org Git - nominatim.git/blob - lib/setup/SetupClass.php
Merge pull request #1667 from mtmail/setup-delete-invalid-indices
[nominatim.git] / lib / setup / SetupClass.php
1 <?php
2
3 namespace Nominatim\Setup;
4
5 require_once(CONST_BasePath.'/lib/setup/AddressLevelParser.php');
6
7 class SetupFunctions
8 {
9     protected $iCacheMemory;
10     protected $iInstances;
11     protected $sModulePath;
12     protected $aDSNInfo;
13     protected $bQuiet;
14     protected $bVerbose;
15     protected $sIgnoreErrors;
16     protected $bEnableDiffUpdates;
17     protected $bEnableDebugStatements;
18     protected $bNoPartitions;
19     protected $oDB = null;
20
21     public function __construct(array $aCMDResult)
22     {
23         // by default, use all but one processor, but never more than 15.
24         $this->iInstances = isset($aCMDResult['threads'])
25             ? $aCMDResult['threads']
26             : (min(16, getProcessorCount()) - 1);
27
28         if ($this->iInstances < 1) {
29             $this->iInstances = 1;
30             warn('resetting threads to '.$this->iInstances);
31         }
32
33         if (isset($aCMDResult['osm2pgsql-cache'])) {
34             $this->iCacheMemory = $aCMDResult['osm2pgsql-cache'];
35         } elseif (!is_null(CONST_Osm2pgsql_Flatnode_File)) {
36             // When flatnode files are enabled then disable cache per default.
37             $this->iCacheMemory = 0;
38         } else {
39             // Otherwise: Assume we can steal all the cache memory in the box.
40             $this->iCacheMemory = getCacheMemoryMB();
41         }
42
43         $this->sModulePath = CONST_Database_Module_Path;
44         info('module path: ' . $this->sModulePath);
45
46         // parse database string
47         $this->aDSNInfo = \Nominatim\DB::parseDSN(CONST_Database_DSN);
48         if (!isset($this->aDSNInfo['port'])) {
49             $this->aDSNInfo['port'] = 5432;
50         }
51
52         // setting member variables based on command line options stored in $aCMDResult
53         $this->bQuiet = $aCMDResult['quiet'];
54         $this->bVerbose = $aCMDResult['verbose'];
55
56         //setting default values which are not set by the update.php array
57         if (isset($aCMDResult['ignore-errors'])) {
58             $this->sIgnoreErrors = $aCMDResult['ignore-errors'];
59         } else {
60             $this->sIgnoreErrors = false;
61         }
62         if (isset($aCMDResult['enable-debug-statements'])) {
63             $this->bEnableDebugStatements = $aCMDResult['enable-debug-statements'];
64         } else {
65             $this->bEnableDebugStatements = false;
66         }
67         if (isset($aCMDResult['no-partitions'])) {
68             $this->bNoPartitions = $aCMDResult['no-partitions'];
69         } else {
70             $this->bNoPartitions = false;
71         }
72         if (isset($aCMDResult['enable-diff-updates'])) {
73             $this->bEnableDiffUpdates = $aCMDResult['enable-diff-updates'];
74         } else {
75             $this->bEnableDiffUpdates = false;
76         }
77     }
78
79     public function createDB()
80     {
81         info('Create DB');
82         $oDB = new \Nominatim\DB;
83
84         if ($oDB->databaseExists()) {
85             fail('database already exists ('.CONST_Database_DSN.')');
86         }
87
88         $sCreateDBCmd = 'createdb -E UTF-8'
89             .' -p '.escapeshellarg($this->aDSNInfo['port'])
90             .' '.escapeshellarg($this->aDSNInfo['database']);
91         if (isset($this->aDSNInfo['username'])) {
92             $sCreateDBCmd .= ' -U '.escapeshellarg($this->aDSNInfo['username']);
93         }
94
95         if (isset($this->aDSNInfo['hostspec'])) {
96             $sCreateDBCmd .= ' -h '.escapeshellarg($this->aDSNInfo['hostspec']);
97         }
98
99         $result = $this->runWithPgEnv($sCreateDBCmd);
100         if ($result != 0) fail('Error executing external command: '.$sCreateDBCmd);
101     }
102
103     public function connect()
104     {
105         $this->oDB = new \Nominatim\DB();
106         $this->oDB->connect();
107     }
108
109     public function setupDB()
110     {
111         info('Setup DB');
112
113         $fPostgresVersion = $this->oDB->getPostgresVersion();
114         echo 'Postgres version found: '.$fPostgresVersion."\n";
115
116         if ($fPostgresVersion < 9.03) {
117             fail('Minimum supported version of Postgresql is 9.3.');
118         }
119
120         $this->pgsqlRunScript('CREATE EXTENSION IF NOT EXISTS hstore');
121         $this->pgsqlRunScript('CREATE EXTENSION IF NOT EXISTS postgis');
122
123         $fPostgisVersion = $this->oDB->getPostgisVersion();
124         echo 'Postgis version found: '.$fPostgisVersion."\n";
125
126         if ($fPostgisVersion < 2.2) {
127             echo "Minimum required Postgis version 2.2\n";
128             exit(1);
129         }
130
131         $i = $this->oDB->getOne("select count(*) from pg_user where usename = '".CONST_Database_Web_User."'");
132         if ($i == 0) {
133             echo "\nERROR: Web user '".CONST_Database_Web_User."' does not exist. Create it with:\n";
134             echo "\n          createuser ".CONST_Database_Web_User."\n\n";
135             exit(1);
136         }
137
138         // Try accessing the C module, so we know early if something is wrong
139         checkModulePresence(); // raises exception on failure
140
141         if (!file_exists(CONST_ExtraDataPath.'/country_osm_grid.sql.gz')) {
142             echo 'Error: you need to download the country_osm_grid first:';
143             echo "\n    wget -O ".CONST_ExtraDataPath."/country_osm_grid.sql.gz https://www.nominatim.org/data/country_grid.sql.gz\n";
144             exit(1);
145         }
146         $this->pgsqlRunScriptFile(CONST_BasePath.'/data/country_name.sql');
147         $this->pgsqlRunScriptFile(CONST_ExtraDataPath.'/country_osm_grid.sql.gz');
148         $this->pgsqlRunScriptFile(CONST_BasePath.'/data/gb_postcode_table.sql');
149         $this->pgsqlRunScriptFile(CONST_BasePath.'/data/us_postcode_table.sql');
150
151         $sPostcodeFilename = CONST_BasePath.'/data/gb_postcode_data.sql.gz';
152         if (file_exists($sPostcodeFilename)) {
153             $this->pgsqlRunScriptFile($sPostcodeFilename);
154         } else {
155             warn('optional external GB postcode table file ('.$sPostcodeFilename.') not found. Skipping.');
156         }
157
158         $sPostcodeFilename = CONST_BasePath.'/data/us_postcode_data.sql.gz';
159         if (file_exists($sPostcodeFilename)) {
160             $this->pgsqlRunScriptFile($sPostcodeFilename);
161         } else {
162             warn('optional external US postcode table file ('.$sPostcodeFilename.') not found. Skipping.');
163         }
164
165         if ($this->bNoPartitions) {
166             $this->pgsqlRunScript('update country_name set partition = 0');
167         }
168     }
169
170     public function importData($sOSMFile)
171     {
172         info('Import data');
173
174         $osm2pgsql = CONST_Osm2pgsql_Binary;
175         if (!file_exists($osm2pgsql)) {
176             echo "Check CONST_Osm2pgsql_Binary in your local settings file.\n";
177             echo "Normally you should not need to set this manually.\n";
178             fail("osm2pgsql not found in '$osm2pgsql'");
179         }
180
181         $osm2pgsql .= ' -S '.escapeshellarg(CONST_Import_Style);
182
183         if (!is_null(CONST_Osm2pgsql_Flatnode_File) && CONST_Osm2pgsql_Flatnode_File) {
184             $osm2pgsql .= ' --flat-nodes '.escapeshellarg(CONST_Osm2pgsql_Flatnode_File);
185         }
186
187         if (CONST_Tablespace_Osm2pgsql_Data)
188             $osm2pgsql .= ' --tablespace-slim-data '.escapeshellarg(CONST_Tablespace_Osm2pgsql_Data);
189         if (CONST_Tablespace_Osm2pgsql_Index)
190             $osm2pgsql .= ' --tablespace-slim-index '.escapeshellarg(CONST_Tablespace_Osm2pgsql_Index);
191         if (CONST_Tablespace_Place_Data)
192             $osm2pgsql .= ' --tablespace-main-data '.escapeshellarg(CONST_Tablespace_Place_Data);
193         if (CONST_Tablespace_Place_Index)
194             $osm2pgsql .= ' --tablespace-main-index '.escapeshellarg(CONST_Tablespace_Place_Index);
195         $osm2pgsql .= ' -lsc -O gazetteer --hstore --number-processes 1';
196         $osm2pgsql .= ' -C '.escapeshellarg($this->iCacheMemory);
197         $osm2pgsql .= ' -P '.escapeshellarg($this->aDSNInfo['port']);
198         if (isset($this->aDSNInfo['username'])) {
199             $osm2pgsql .= ' -U '.escapeshellarg($this->aDSNInfo['username']);
200         }
201         if (isset($this->aDSNInfo['hostspec'])) {
202             $osm2pgsql .= ' -H '.escapeshellarg($this->aDSNInfo['hostspec']);
203         }
204         $osm2pgsql .= ' -d '.escapeshellarg($this->aDSNInfo['database']).' '.escapeshellarg($sOSMFile);
205
206         $this->runWithPgEnv($osm2pgsql);
207
208         if (!$this->sIgnoreErrors && !$this->oDB->getRow('select * from place limit 1')) {
209             fail('No Data');
210         }
211     }
212
213     public function createFunctions()
214     {
215         info('Create Functions');
216
217         // Try accessing the C module, so we know early if something is wrong
218         checkModulePresence(); // raises exception on failure
219
220         $this->createSqlFunctions();
221     }
222
223     public function createTables($bReverseOnly = false)
224     {
225         info('Create Tables');
226
227         $sTemplate = file_get_contents(CONST_BasePath.'/sql/tables.sql');
228         $sTemplate = $this->replaceSqlPatterns($sTemplate);
229
230         $this->pgsqlRunScript($sTemplate, false);
231
232         if ($bReverseOnly) {
233             $this->dropTable('search_name');
234         }
235
236         $oAlParser = new AddressLevelParser(CONST_Address_Level_Config);
237         $oAlParser->createTable($this->oDB, 'address_levels');
238     }
239
240     public function createTableTriggers()
241     {
242         info('Create Tables');
243
244         $sTemplate = file_get_contents(CONST_BasePath.'/sql/table-triggers.sql');
245         $sTemplate = $this->replaceSqlPatterns($sTemplate);
246
247         $this->pgsqlRunScript($sTemplate, false);
248     }
249
250     public function createPartitionTables()
251     {
252         info('Create Partition Tables');
253
254         $sTemplate = file_get_contents(CONST_BasePath.'/sql/partition-tables.src.sql');
255         $sTemplate = $this->replaceSqlPatterns($sTemplate);
256
257         $this->pgsqlRunPartitionScript($sTemplate);
258     }
259
260     public function createPartitionFunctions()
261     {
262         info('Create Partition Functions');
263
264         $sTemplate = file_get_contents(CONST_BasePath.'/sql/partition-functions.src.sql');
265         $this->pgsqlRunPartitionScript($sTemplate);
266     }
267
268     public function importWikipediaArticles()
269     {
270         $sWikiArticlesFile = CONST_Wikipedia_Data_Path.'/wikimedia-importance.sql.gz';
271         if (file_exists($sWikiArticlesFile)) {
272             info('Importing wikipedia articles and redirects');
273             $this->dropTable('wikipedia_article');
274             $this->dropTable('wikipedia_redirect');
275             $this->pgsqlRunScriptFile($sWikiArticlesFile);
276         } else {
277             warn('wikipedia importance dump file not found - places will have default importance');
278         }
279     }
280
281     public function loadData($bDisableTokenPrecalc)
282     {
283         info('Drop old Data');
284
285         $this->oDB->exec('TRUNCATE word');
286         echo '.';
287         $this->oDB->exec('TRUNCATE placex');
288         echo '.';
289         $this->oDB->exec('TRUNCATE location_property_osmline');
290         echo '.';
291         $this->oDB->exec('TRUNCATE place_addressline');
292         echo '.';
293         $this->oDB->exec('TRUNCATE location_area');
294         echo '.';
295         if (!$this->dbReverseOnly()) {
296             $this->oDB->exec('TRUNCATE search_name');
297             echo '.';
298         }
299         $this->oDB->exec('TRUNCATE search_name_blank');
300         echo '.';
301         $this->oDB->exec('DROP SEQUENCE seq_place');
302         echo '.';
303         $this->oDB->exec('CREATE SEQUENCE seq_place start 100000');
304         echo '.';
305
306         $sSQL = 'select distinct partition from country_name';
307         $aPartitions = $this->oDB->getCol($sSQL);
308
309         if (!$this->bNoPartitions) $aPartitions[] = 0;
310         foreach ($aPartitions as $sPartition) {
311             $this->oDB->exec('TRUNCATE location_road_'.$sPartition);
312             echo '.';
313         }
314
315         // used by getorcreate_word_id to ignore frequent partial words
316         $sSQL = 'CREATE OR REPLACE FUNCTION get_maxwordfreq() RETURNS integer AS ';
317         $sSQL .= '$$ SELECT '.CONST_Max_Word_Frequency.' as maxwordfreq; $$ LANGUAGE SQL IMMUTABLE';
318         $this->oDB->exec($sSQL);
319         echo ".\n";
320
321         // pre-create the word list
322         if (!$bDisableTokenPrecalc) {
323             info('Loading word list');
324             $this->pgsqlRunScriptFile(CONST_BasePath.'/data/words.sql');
325         }
326
327         info('Load Data');
328         $sColumns = 'osm_type, osm_id, class, type, name, admin_level, address, extratags, geometry';
329
330         $aDBInstances = array();
331         $iLoadThreads = max(1, $this->iInstances - 1);
332         for ($i = 0; $i < $iLoadThreads; $i++) {
333             // https://secure.php.net/manual/en/function.pg-connect.php
334             $DSN = CONST_Database_DSN;
335             $DSN = preg_replace('/^pgsql:/', '', $DSN);
336             $DSN = preg_replace('/;/', ' ', $DSN);
337             $aDBInstances[$i] = pg_connect($DSN, PGSQL_CONNECT_FORCE_NEW);
338             pg_ping($aDBInstances[$i]);
339         }
340
341         for ($i = 0; $i < $iLoadThreads; $i++) {
342             $sSQL = "INSERT INTO placex ($sColumns) SELECT $sColumns FROM place WHERE osm_id % $iLoadThreads = $i";
343             $sSQL .= " and not (class='place' and type='houses' and osm_type='W'";
344             $sSQL .= "          and ST_GeometryType(geometry) = 'ST_LineString')";
345             $sSQL .= ' and ST_IsValid(geometry)';
346             if ($this->bVerbose) echo "$sSQL\n";
347             if (!pg_send_query($aDBInstances[$i], $sSQL)) {
348                 fail(pg_last_error($aDBInstances[$i]));
349             }
350         }
351
352         // last thread for interpolation lines
353         // https://secure.php.net/manual/en/function.pg-connect.php
354         $DSN = CONST_Database_DSN;
355         $DSN = preg_replace('/^pgsql:/', '', $DSN);
356         $DSN = preg_replace('/;/', ' ', $DSN);
357         $aDBInstances[$iLoadThreads] = pg_connect($DSN, PGSQL_CONNECT_FORCE_NEW);
358         pg_ping($aDBInstances[$iLoadThreads]);
359         $sSQL = 'insert into location_property_osmline';
360         $sSQL .= ' (osm_id, address, linegeo)';
361         $sSQL .= ' SELECT osm_id, address, geometry from place where ';
362         $sSQL .= "class='place' and type='houses' and osm_type='W' and ST_GeometryType(geometry) = 'ST_LineString'";
363         if ($this->bVerbose) echo "$sSQL\n";
364         if (!pg_send_query($aDBInstances[$iLoadThreads], $sSQL)) {
365             fail(pg_last_error($aDBInstances[$iLoadThreads]));
366         }
367
368         $bFailed = false;
369         for ($i = 0; $i <= $iLoadThreads; $i++) {
370             while (($hPGresult = pg_get_result($aDBInstances[$i])) !== false) {
371                 $resultStatus = pg_result_status($hPGresult);
372                 // PGSQL_EMPTY_QUERY, PGSQL_COMMAND_OK, PGSQL_TUPLES_OK,
373                 // PGSQL_COPY_OUT, PGSQL_COPY_IN, PGSQL_BAD_RESPONSE,
374                 // PGSQL_NONFATAL_ERROR and PGSQL_FATAL_ERROR
375                 // echo 'Query result ' . $i . ' is: ' . $resultStatus . "\n";
376                 if ($resultStatus != PGSQL_COMMAND_OK && $resultStatus != PGSQL_TUPLES_OK) {
377                     $resultError = pg_result_error($hPGresult);
378                     echo '-- error text ' . $i . ': ' . $resultError . "\n";
379                     $bFailed = true;
380                 }
381             }
382         }
383         if ($bFailed) {
384             fail('SQL errors loading placex and/or location_property_osmline tables');
385         }
386
387         for ($i = 0; $i < $this->iInstances; $i++) {
388             pg_close($aDBInstances[$i]);
389         }
390
391         echo "\n";
392         info('Reanalysing database');
393         $this->pgsqlRunScript('ANALYSE');
394
395         $sDatabaseDate = getDatabaseDate($this->oDB);
396         $this->oDB->exec('TRUNCATE import_status');
397         if (!$sDatabaseDate) {
398             warn('could not determine database date.');
399         } else {
400             $sSQL = "INSERT INTO import_status (lastimportdate) VALUES('".$sDatabaseDate."')";
401             $this->oDB->exec($sSQL);
402             echo "Latest data imported from $sDatabaseDate.\n";
403         }
404     }
405
406     public function importTigerData()
407     {
408         info('Import Tiger data');
409
410         $aFilenames = glob(CONST_Tiger_Data_Path.'/*.sql');
411         info('Found '.count($aFilenames).' SQL files in path '.CONST_Tiger_Data_Path);
412         if (empty($aFilenames)) return;
413
414         $sTemplate = file_get_contents(CONST_BasePath.'/sql/tiger_import_start.sql');
415         $sTemplate = $this->replaceSqlPatterns($sTemplate);
416
417         $this->pgsqlRunScript($sTemplate, false);
418
419         $aDBInstances = array();
420         for ($i = 0; $i < $this->iInstances; $i++) {
421             // https://secure.php.net/manual/en/function.pg-connect.php
422             $DSN = CONST_Database_DSN;
423             $DSN = preg_replace('/^pgsql:/', '', $DSN);
424             $DSN = preg_replace('/;/', ' ', $DSN);
425             $aDBInstances[$i] = pg_connect($DSN, PGSQL_CONNECT_FORCE_NEW | PGSQL_CONNECT_ASYNC);
426             pg_ping($aDBInstances[$i]);
427         }
428
429         foreach ($aFilenames as $sFile) {
430             echo $sFile.': ';
431             $hFile = fopen($sFile, 'r');
432             $sSQL = fgets($hFile, 100000);
433             $iLines = 0;
434             while (true) {
435                 for ($i = 0; $i < $this->iInstances; $i++) {
436                     if (!pg_connection_busy($aDBInstances[$i])) {
437                         while (pg_get_result($aDBInstances[$i]));
438                         $sSQL = fgets($hFile, 100000);
439                         if (!$sSQL) break 2;
440                         if (!pg_send_query($aDBInstances[$i], $sSQL)) fail(pg_last_error($aDBInstances[$i]));
441                         $iLines++;
442                         if ($iLines == 1000) {
443                             echo '.';
444                             $iLines = 0;
445                         }
446                     }
447                 }
448                 usleep(10);
449             }
450             fclose($hFile);
451
452             $bAnyBusy = true;
453             while ($bAnyBusy) {
454                 $bAnyBusy = false;
455                 for ($i = 0; $i < $this->iInstances; $i++) {
456                     if (pg_connection_busy($aDBInstances[$i])) $bAnyBusy = true;
457                 }
458                 usleep(10);
459             }
460             echo "\n";
461         }
462
463         for ($i = 0; $i < $this->iInstances; $i++) {
464             pg_close($aDBInstances[$i]);
465         }
466
467         info('Creating indexes on Tiger data');
468         $sTemplate = file_get_contents(CONST_BasePath.'/sql/tiger_import_finish.sql');
469         $sTemplate = $this->replaceSqlPatterns($sTemplate);
470
471         $this->pgsqlRunScript($sTemplate, false);
472     }
473
474     public function calculatePostcodes($bCMDResultAll)
475     {
476         info('Calculate Postcodes');
477         $this->oDB->exec('TRUNCATE location_postcode');
478
479         $sSQL  = 'INSERT INTO location_postcode';
480         $sSQL .= ' (place_id, indexed_status, country_code, postcode, geometry) ';
481         $sSQL .= "SELECT nextval('seq_place'), 1, country_code,";
482         $sSQL .= "       upper(trim (both ' ' from address->'postcode')) as pc,";
483         $sSQL .= '       ST_Centroid(ST_Collect(ST_Centroid(geometry)))';
484         $sSQL .= '  FROM placex';
485         $sSQL .= " WHERE address ? 'postcode' AND address->'postcode' NOT SIMILAR TO '%(,|;)%'";
486         $sSQL .= '       AND geometry IS NOT null';
487         $sSQL .= ' GROUP BY country_code, pc';
488         $this->oDB->exec($sSQL);
489
490         // only add postcodes that are not yet available in OSM
491         $sSQL  = 'INSERT INTO location_postcode';
492         $sSQL .= ' (place_id, indexed_status, country_code, postcode, geometry) ';
493         $sSQL .= "SELECT nextval('seq_place'), 1, 'us', postcode,";
494         $sSQL .= '       ST_SetSRID(ST_Point(x,y),4326)';
495         $sSQL .= '  FROM us_postcode WHERE postcode NOT IN';
496         $sSQL .= '        (SELECT postcode FROM location_postcode';
497         $sSQL .= "          WHERE country_code = 'us')";
498         $this->oDB->exec($sSQL);
499
500         // add missing postcodes for GB (if available)
501         $sSQL  = 'INSERT INTO location_postcode';
502         $sSQL .= ' (place_id, indexed_status, country_code, postcode, geometry) ';
503         $sSQL .= "SELECT nextval('seq_place'), 1, 'gb', postcode, geometry";
504         $sSQL .= '  FROM gb_postcode WHERE postcode NOT IN';
505         $sSQL .= '           (SELECT postcode FROM location_postcode';
506         $sSQL .= "             WHERE country_code = 'gb')";
507         $this->oDB->exec($sSQL);
508
509         if (!$bCMDResultAll) {
510             $sSQL = "DELETE FROM word WHERE class='place' and type='postcode'";
511             $sSQL .= 'and word NOT IN (SELECT postcode FROM location_postcode)';
512             $this->oDB->exec($sSQL);
513         }
514
515         $sSQL = 'SELECT count(getorcreate_postcode_id(v)) FROM ';
516         $sSQL .= '(SELECT distinct(postcode) as v FROM location_postcode) p';
517         $this->oDB->exec($sSQL);
518     }
519
520     public function index($bIndexNoanalyse)
521     {
522         $sOutputFile = '';
523         $sBaseCmd = CONST_BasePath.'/nominatim/nominatim.py'
524             .' -d '.escapeshellarg($this->aDSNInfo['database'])
525             .' -P '.escapeshellarg($this->aDSNInfo['port'])
526             .' -t '.escapeshellarg($this->iInstances.$sOutputFile);
527         if (!$this->bQuiet) {
528             $sBaseCmd .= ' -v';
529         }
530         if ($this->bVerbose) {
531             $sBaseCmd .= ' -v';
532         }
533         if (isset($this->aDSNInfo['hostspec'])) {
534             $sBaseCmd .= ' -H '.escapeshellarg($this->aDSNInfo['hostspec']);
535         }
536         if (isset($this->aDSNInfo['username'])) {
537             $sBaseCmd .= ' -U '.escapeshellarg($this->aDSNInfo['username']);
538         }
539
540         info('Index ranks 0 - 4');
541         $iStatus = $this->runWithPgEnv($sBaseCmd.' -R 4');
542         if ($iStatus != 0) {
543             fail('error status ' . $iStatus . ' running nominatim!');
544         }
545         if (!$bIndexNoanalyse) $this->pgsqlRunScript('ANALYSE');
546
547         info('Index ranks 5 - 25');
548         $iStatus = $this->runWithPgEnv($sBaseCmd.' -r 5 -R 25');
549         if ($iStatus != 0) {
550             fail('error status ' . $iStatus . ' running nominatim!');
551         }
552         if (!$bIndexNoanalyse) $this->pgsqlRunScript('ANALYSE');
553
554         info('Index ranks 26 - 30');
555         $iStatus = $this->runWithPgEnv($sBaseCmd.' -r 26');
556         if ($iStatus != 0) {
557             fail('error status ' . $iStatus . ' running nominatim!');
558         }
559
560         info('Index postcodes');
561         $sSQL = 'UPDATE location_postcode SET indexed_status = 0';
562         $this->oDB->exec($sSQL);
563     }
564
565     public function createSearchIndices()
566     {
567         info('Create Search indices');
568
569         $sSQL = 'SELECT relname FROM pg_class, pg_index ';
570         $sSQL .= 'WHERE pg_index.indisvalid = false AND pg_index.indexrelid = pg_class.oid';
571         $aInvalidIndices = $this->oDB->getCol($sSQL);
572
573         foreach ($aInvalidIndices as $sIndexName) {
574             info("Cleaning up invalid index $sIndexName");
575             $this->oDB->exec("DROP INDEX $sIndexName;");
576         }
577
578         $sTemplate = file_get_contents(CONST_BasePath.'/sql/indices.src.sql');
579         if (!$this->dbReverseOnly()) {
580             $sTemplate .= file_get_contents(CONST_BasePath.'/sql/indices_search.src.sql');
581         }
582         $sTemplate = $this->replaceSqlPatterns($sTemplate);
583
584         $this->pgsqlRunScript($sTemplate);
585     }
586
587     public function createCountryNames()
588     {
589         info('Create search index for default country names');
590
591         $this->pgsqlRunScript("select getorcreate_country(make_standard_name('uk'), 'gb')");
592         $this->pgsqlRunScript("select getorcreate_country(make_standard_name('united states'), 'us')");
593         $this->pgsqlRunScript('select count(*) from (select getorcreate_country(make_standard_name(country_code), country_code) from country_name where country_code is not null) as x');
594         $this->pgsqlRunScript("select count(*) from (select getorcreate_country(make_standard_name(name->'name'), country_code) from country_name where name ? 'name') as x");
595         $sSQL = 'select count(*) from (select getorcreate_country(make_standard_name(v),'
596             .'country_code) from (select country_code, skeys(name) as k, svals(name) as v from country_name) x where k ';
597         if (CONST_Languages) {
598             $sSQL .= 'in ';
599             $sDelim = '(';
600             foreach (explode(',', CONST_Languages) as $sLang) {
601                 $sSQL .= $sDelim."'name:$sLang'";
602                 $sDelim = ',';
603             }
604             $sSQL .= ')';
605         } else {
606             // all include all simple name tags
607             $sSQL .= "like 'name:%'";
608         }
609         $sSQL .= ') v';
610         $this->pgsqlRunScript($sSQL);
611     }
612
613     public function drop()
614     {
615         info('Drop tables only required for updates');
616
617         // The implementation is potentially a bit dangerous because it uses
618         // a positive selection of tables to keep, and deletes everything else.
619         // Including any tables that the unsuspecting user might have manually
620         // created. USE AT YOUR OWN PERIL.
621         // tables we want to keep. everything else goes.
622         $aKeepTables = array(
623                         '*columns',
624                         'import_polygon_*',
625                         'import_status',
626                         'place_addressline',
627                         'location_postcode',
628                         'location_property*',
629                         'placex',
630                         'search_name',
631                         'seq_*',
632                         'word',
633                         'query_log',
634                         'new_query_log',
635                         'spatial_ref_sys',
636                         'country_name',
637                         'place_classtype_*',
638                         'country_osm_grid'
639                        );
640
641         $aDropTables = array();
642         $aHaveTables = $this->oDB->getCol("SELECT tablename FROM pg_tables WHERE schemaname='public'");
643
644         foreach ($aHaveTables as $sTable) {
645             $bFound = false;
646             foreach ($aKeepTables as $sKeep) {
647                 if (fnmatch($sKeep, $sTable)) {
648                     $bFound = true;
649                     break;
650                 }
651             }
652             if (!$bFound) array_push($aDropTables, $sTable);
653         }
654         foreach ($aDropTables as $sDrop) {
655             $this->dropTable($sDrop);
656         }
657
658         if (!is_null(CONST_Osm2pgsql_Flatnode_File) && CONST_Osm2pgsql_Flatnode_File) {
659             if (file_exists(CONST_Osm2pgsql_Flatnode_File)) {
660                 if ($this->bVerbose) echo 'Deleting '.CONST_Osm2pgsql_Flatnode_File."\n";
661                 unlink(CONST_Osm2pgsql_Flatnode_File);
662             }
663         }
664     }
665
666     private function pgsqlRunScript($sScript, $bfatal = true)
667     {
668         runSQLScript(
669             $sScript,
670             $bfatal,
671             $this->bVerbose,
672             $this->sIgnoreErrors
673         );
674     }
675
676     private function createSqlFunctions()
677     {
678         $sBasePath = CONST_BasePath.'/sql/functions/';
679         $sTemplate = file_get_contents($sBasePath.'utils.sql');
680         $sTemplate .= file_get_contents($sBasePath.'normalization.sql');
681         $sTemplate .= file_get_contents($sBasePath.'importance.sql');
682         $sTemplate .= file_get_contents($sBasePath.'address_lookup.sql');
683         $sTemplate .= file_get_contents($sBasePath.'interpolation.sql');
684         if ($this->oDB->tableExists('place')) {
685             $sTemplate .= file_get_contents($sBasePath.'place_triggers.sql');
686         }
687         if ($this->oDB->tableExists('placex')) {
688             $sTemplate .= file_get_contents($sBasePath.'placex_triggers.sql');
689         }
690         if ($this->oDB->tableExists('location_postcode')) {
691             $sTemplate .= file_get_contents($sBasePath.'postcode_triggers.sql');
692         }
693         $sTemplate = str_replace('{modulepath}', $this->sModulePath, $sTemplate);
694         if ($this->bEnableDiffUpdates) {
695             $sTemplate = str_replace('RETURN NEW; -- %DIFFUPDATES%', '--', $sTemplate);
696         }
697         if ($this->bEnableDebugStatements) {
698             $sTemplate = str_replace('--DEBUG:', '', $sTemplate);
699         }
700         if (CONST_Limit_Reindexing) {
701             $sTemplate = str_replace('--LIMIT INDEXING:', '', $sTemplate);
702         }
703         if (!CONST_Use_US_Tiger_Data) {
704             $sTemplate = str_replace('-- %NOTIGERDATA% ', '', $sTemplate);
705         }
706         if (!CONST_Use_Aux_Location_data) {
707             $sTemplate = str_replace('-- %NOAUXDATA% ', '', $sTemplate);
708         }
709
710         $sReverseOnly = $this->dbReverseOnly() ? 'true' : 'false';
711         $sTemplate = str_replace('%REVERSE-ONLY%', $sReverseOnly, $sTemplate);
712
713         $this->pgsqlRunScript($sTemplate);
714     }
715
716     private function pgsqlRunPartitionScript($sTemplate)
717     {
718         $sSQL = 'select distinct partition from country_name';
719         $aPartitions = $this->oDB->getCol($sSQL);
720         if (!$this->bNoPartitions) $aPartitions[] = 0;
721
722         preg_match_all('#^-- start(.*?)^-- end#ms', $sTemplate, $aMatches, PREG_SET_ORDER);
723         foreach ($aMatches as $aMatch) {
724             $sResult = '';
725             foreach ($aPartitions as $sPartitionName) {
726                 $sResult .= str_replace('-partition-', $sPartitionName, $aMatch[1]);
727             }
728             $sTemplate = str_replace($aMatch[0], $sResult, $sTemplate);
729         }
730
731         $this->pgsqlRunScript($sTemplate);
732     }
733
734     private function pgsqlRunScriptFile($sFilename)
735     {
736         if (!file_exists($sFilename)) fail('unable to find '.$sFilename);
737
738         $sCMD = 'psql'
739             .' -p '.escapeshellarg($this->aDSNInfo['port'])
740             .' -d '.escapeshellarg($this->aDSNInfo['database']);
741         if (!$this->bVerbose) {
742             $sCMD .= ' -q';
743         }
744         if (isset($this->aDSNInfo['hostspec'])) {
745             $sCMD .= ' -h '.escapeshellarg($this->aDSNInfo['hostspec']);
746         }
747         if (isset($this->aDSNInfo['username'])) {
748             $sCMD .= ' -U '.escapeshellarg($this->aDSNInfo['username']);
749         }
750         $aProcEnv = null;
751         if (isset($this->aDSNInfo['password'])) {
752             $aProcEnv = array_merge(array('PGPASSWORD' => $this->aDSNInfo['password']), $_ENV);
753         }
754         $ahGzipPipes = null;
755         if (preg_match('/\\.gz$/', $sFilename)) {
756             $aDescriptors = array(
757                              0 => array('pipe', 'r'),
758                              1 => array('pipe', 'w'),
759                              2 => array('file', '/dev/null', 'a')
760                             );
761             $hGzipProcess = proc_open('zcat '.escapeshellarg($sFilename), $aDescriptors, $ahGzipPipes);
762             if (!is_resource($hGzipProcess)) fail('unable to start zcat');
763             $aReadPipe = $ahGzipPipes[1];
764             fclose($ahGzipPipes[0]);
765         } else {
766             $sCMD .= ' -f '.escapeshellarg($sFilename);
767             $aReadPipe = array('pipe', 'r');
768         }
769         $aDescriptors = array(
770                          0 => $aReadPipe,
771                          1 => array('pipe', 'w'),
772                          2 => array('file', '/dev/null', 'a')
773                         );
774         $ahPipes = null;
775         $hProcess = proc_open($sCMD, $aDescriptors, $ahPipes, null, $aProcEnv);
776         if (!is_resource($hProcess)) fail('unable to start pgsql');
777         // TODO: error checking
778         while (!feof($ahPipes[1])) {
779             echo fread($ahPipes[1], 4096);
780         }
781         fclose($ahPipes[1]);
782         $iReturn = proc_close($hProcess);
783         if ($iReturn > 0) {
784             fail("pgsql returned with error code ($iReturn)");
785         }
786         if ($ahGzipPipes) {
787             fclose($ahGzipPipes[1]);
788             proc_close($hGzipProcess);
789         }
790     }
791
792     private function replaceSqlPatterns($sSql)
793     {
794         $sSql = str_replace('{www-user}', CONST_Database_Web_User, $sSql);
795
796         $aPatterns = array(
797                       '{ts:address-data}' => CONST_Tablespace_Address_Data,
798                       '{ts:address-index}' => CONST_Tablespace_Address_Index,
799                       '{ts:search-data}' => CONST_Tablespace_Search_Data,
800                       '{ts:search-index}' =>  CONST_Tablespace_Search_Index,
801                       '{ts:aux-data}' =>  CONST_Tablespace_Aux_Data,
802                       '{ts:aux-index}' =>  CONST_Tablespace_Aux_Index,
803         );
804
805         foreach ($aPatterns as $sPattern => $sTablespace) {
806             if ($sTablespace) {
807                 $sSql = str_replace($sPattern, 'TABLESPACE "'.$sTablespace.'"', $sSql);
808             } else {
809                 $sSql = str_replace($sPattern, '', $sSql);
810             }
811         }
812
813         return $sSql;
814     }
815
816     private function runWithPgEnv($sCmd)
817     {
818         if ($this->bVerbose) {
819             echo "Execute: $sCmd\n";
820         }
821
822         $aProcEnv = null;
823
824         if (isset($this->aDSNInfo['password'])) {
825             $aProcEnv = array_merge(array('PGPASSWORD' => $this->aDSNInfo['password']), $_ENV);
826         }
827
828         return runWithEnv($sCmd, $aProcEnv);
829     }
830
831     /**
832      * Drop table with the given name if it exists.
833      *
834      * @param string $sName Name of table to remove.
835      *
836      * @return null
837      *
838      * @pre connect() must have been called.
839      */
840     private function dropTable($sName)
841     {
842         if ($this->bVerbose) echo "Dropping table $sName\n";
843         $this->oDB->exec('DROP TABLE IF EXISTS '.$sName.' CASCADE');
844     }
845
846     /**
847      * Check if the database is in reverse-only mode.
848      *
849      * @return True if there is no search_name table and infrastructure.
850      */
851     private function dbReverseOnly()
852     {
853         return !($this->oDB->tableExists('search_name'));
854     }
855 }