includes/db/DatabasePostgres.php

   1 <?php
   2 /**
   3  * This is the Postgres database abstraction layer.
   4  *
   5  * @file
   6  * @ingroup Database
   7  */
   8
   9 class PostgresField implements Field {
  10         private $name, $tablename, $type, $nullable, $max_length, $deferred, $deferrable, $conname;
  11
  12         static function fromText($db, $table, $field) {
  13         global $wgDBmwschema;
  14
  15                 $q = <<<SQL
  16 SELECT
  17  attnotnull, attlen, COALESCE(conname, '') AS conname,
  18  COALESCE(condeferred, 'f') AS deferred,
  19  COALESCE(condeferrable, 'f') AS deferrable,
  20  CASE WHEN typname = 'int2' THEN 'smallint'
  21   WHEN typname = 'int4' THEN 'integer'
  22   WHEN typname = 'int8' THEN 'bigint'
  23   WHEN typname = 'bpchar' THEN 'char'
  24  ELSE typname END AS typname
  25 FROM pg_class c
  26 JOIN pg_namespace n ON (n.oid = c.relnamespace)
  27 JOIN pg_attribute a ON (a.attrelid = c.oid)
  28 JOIN pg_type t ON (t.oid = a.atttypid)
  29 LEFT JOIN pg_constraint o ON (o.conrelid = c.oid AND a.attnum = ANY(o.conkey) AND o.contype = 'f')
  30 WHERE relkind = 'r'
  31 AND nspname=%s
  32 AND relname=%s
  33 AND attname=%s;
  34 SQL;
  35
  36                 $table = $db->tableName( $table );
  37                 $res = $db->query(
  38                         sprintf( $q,
  39                                 $db->addQuotes( $wgDBmwschema ),
  40                                 $db->addQuotes( $table ),
  41                                 $db->addQuotes( $field )
  42                         )
  43                 );
  44                 $row = $db->fetchObject( $res );
  45                 if ( !$row ) {
  46                         return null;
  47                 }
  48                 $n = new PostgresField;
  49                 $n->type = $row->typname;
  50                 $n->nullable = ( $row->attnotnull == 'f' );
  51                 $n->name = $field;
  52                 $n->tablename = $table;
  53                 $n->max_length = $row->attlen;
  54                 $n->deferrable = ( $row->deferrable == 't' );
  55                 $n->deferred = ( $row->deferred == 't' );
  56                 $n->conname = $row->conname;
  57                 return $n;
  58         }
  59
  60         function name() {
  61                 return $this->name;
  62         }
  63
  64         function tableName() {
  65                 return $this->tablename;
  66         }
  67
  68         function type() {
  69                 return $this->type;
  70         }
  71
  72         function isNullable() {
  73                 return $this->nullable;
  74         }
  75
  76         function maxLength() {
  77                 return $this->max_length;
  78         }
  79
  80         function is_deferrable() {
  81                 return $this->deferrable;
  82         }
  83
  84         function is_deferred() {
  85                 return $this->deferred;
  86         }
  87
  88         function conname() {
  89                 return $this->conname;
  90         }
  91
  92 }
  93
  94 /**
  95  * @ingroup Database
  96  */
  97 class DatabasePostgres extends DatabaseBase {
  98         var $mInsertId = null;
  99         var $mLastResult = null;
 100         var $numeric_version = null;
 101         var $mAffectedRows = null;
 102
 103         function getType() {
 104                 return 'postgres';
 105         }
 106
 107         function cascadingDeletes() {
 108                 return true;
 109         }
 110         function cleanupTriggers() {
 111                 return true;
 112         }
 113         function strictIPs() {
 114                 return true;
 115         }
 116         function realTimestamps() {
 117                 return true;
 118         }
 119         function implicitGroupby() {
 120                 return false;
 121         }
 122         function implicitOrderby() {
 123                 return false;
 124         }
 125         function searchableIPs() {
 126                 return true;
 127         }
 128         function functionalIndexes() {
 129                 return true;
 130         }
 131
 132         function hasConstraint( $name ) {
 133                 global $wgDBmwschema;
 134                 $SQL = "SELECT 1 FROM pg_catalog.pg_constraint c, pg_catalog.pg_namespace n WHERE c.connamespace = n.oid AND conname = '" .
 135                                 pg_escape_string( $this->mConn, $name ) . "' AND n.nspname = '" . pg_escape_string( $this->mConn, $wgDBmwschema ) ."'";
 136                 $res = $this->doQuery( $SQL );
 137                 return $this->numRows( $res );
 138         }
 139
 140         /**
 141          * Usually aborts on failure
 142          */
 143         function open( $server, $user, $password, $dbName ) {
 144                 # Test for Postgres support, to avoid suppressed fatal error
 145                 if ( !function_exists( 'pg_connect' ) ) {
 146                         throw new DBConnectionError( $this, "Postgres functions missing, have you compiled PHP with the --with-pgsql option?\n (Note: if you recently installed PHP, you may need to restart your webserver and database)\n" );
 147                 }
 148
 149                 global $wgDBport;
 150
 151                 if ( !strlen( $user ) ) { # e.g. the class is being loaded
 152                         return;
 153                 }
 154                 $this->close();
 155                 $this->mServer = $server;
 156                 $this->mPort = $port = $wgDBport;
 157                 $this->mUser = $user;
 158                 $this->mPassword = $password;
 159                 $this->mDBname = $dbName;
 160
 161                 $connectVars = array(
 162                         'dbname' => $dbName,
 163                         'user' => $user,
 164                         'password' => $password
 165                 );
 166                 if ( $server != false && $server != '' ) {
 167                         $connectVars['host'] = $server;
 168                 }
 169                 if ( $port != false && $port != '' ) {
 170                         $connectVars['port'] = $port;
 171                 }
 172                 $connectString = $this->makeConnectionString( $connectVars, PGSQL_CONNECT_FORCE_NEW );
 173
 174                 $this->installErrorHandler();
 175                 $this->mConn = pg_connect( $connectString );
 176                 $phpError = $this->restoreErrorHandler();
 177
 178                 if ( !$this->mConn ) {
 179                         wfDebug( "DB connection error\n" );
 180                         wfDebug( "Server: $server, Database: $dbName, User: $user, Password: " . substr( $password, 0, 3 ) . "...\n" );
 181                         wfDebug( $this->lastError() . "\n" );
 182                         throw new DBConnectionError( $this, $phpError );
 183                 }
 184
 185                 $this->mOpened = true;
 186
 187                 global $wgCommandLineMode;
 188                 # If called from the command-line (e.g. importDump), only show errors
 189                 if ( $wgCommandLineMode ) {
 190                         $this->doQuery( "SET client_min_messages = 'ERROR'" );
 191                 }
 192
 193                 $this->doQuery( "SET client_encoding='UTF8'" );
 194
 195                 global $wgDBmwschema;
 196                 if ( isset( $wgDBmwschema )
 197                         && $wgDBmwschema !== 'mediawiki'
 198                         && preg_match( '/^\w+$/', $wgDBmwschema )
 199                 ) {
 200                         $safeschema = $this->addIdentifierQuotes( $wgDBmwschema );
 201                         $this->doQuery( "SET search_path = $safeschema, public" );
 202                 }
 203
 204                 return $this->mConn;
 205         }
 206
 207         function makeConnectionString( $vars ) {
 208                 $s = '';
 209                 foreach ( $vars as $name => $value ) {
 210                         $s .= "$name='" . str_replace( "'", "\\'", $value ) . "' ";
 211                 }
 212                 return $s;
 213         }
 214
 215
 216         function initial_setup( $superuser, $password, $dbName ) {
 217                 // If this is the initial connection, setup the schema stuff and possibly create the user
 218                 global $wgDBname, $wgDBuser, $wgDBpassword, $wgDBmwschema, $wgDBts2schema;
 219
 220                 $safeuser = $this->addIdentifierQuotes( $wgDBuser );
 221                 // Are we connecting as a superuser for the first time?
 222                 if ( $superuser ) {
 223                         // Are we really a superuser? Check out our rights
 224                         $SQL = "SELECT
 225                       CASE WHEN usesuper IS TRUE THEN
 226                       CASE WHEN usecreatedb IS TRUE THEN 3 ELSE 1 END
 227                       ELSE CASE WHEN usecreatedb IS TRUE THEN 2 ELSE 0 END
 228                     END AS rights
 229                     FROM pg_catalog.pg_user WHERE usename = " . $this->addQuotes( $superuser );
 230                         $rows = $this->numRows( $res = $this->doQuery( $SQL ) );
 231                         if ( !$rows ) {
 232                                 print '<li>ERROR: Could not read permissions for user "' . htmlspecialchars( $superuser ) . "\"</li>\n";
 233                                 dieout( );
 234                         }
 235                         $perms = pg_fetch_result( $res, 0, 0 );
 236
 237                         $SQL = "SELECT 1 FROM pg_catalog.pg_user WHERE usename = " . $this->addQuotes( $wgDBuser );
 238                         $rows = $this->numRows( $this->doQuery( $SQL ) );
 239                         if ( $rows ) {
 240                                 print '<li>User "' . htmlspecialchars( $wgDBuser ) . '" already exists, skipping account creation.</li>';
 241                         } else {
 242                                 if ( $perms != 1 && $perms != 3 ) {
 243                                         print '<li>ERROR: the user "' . htmlspecialchars( $superuser ) . '" cannot create other users. ';
 244                                         print 'Please use a different Postgres user.</li>';
 245                                         dieout( );
 246                                 }
 247                                 print '<li>Creating user <b>' . htmlspecialchars( $wgDBuser ) . '</b>...';
 248                                 $safepass = $this->addQuotes( $wgDBpassword );
 249                                 $SQL = "CREATE USER $safeuser NOCREATEDB PASSWORD $safepass";
 250                                 $this->doQuery( $SQL );
 251                                 print "OK</li>\n";
 252                         }
 253                         // User now exists, check out the database
 254                         if ( $dbName != $wgDBname ) {
 255                                 $SQL = "SELECT 1 FROM pg_catalog.pg_database WHERE datname = " . $this->addQuotes( $wgDBname );
 256                                 $rows = $this->numRows( $this->doQuery( $SQL ) );
 257                                 if ( $rows ) {
 258                                         print '<li>Database "' . htmlspecialchars( $wgDBname ) . '" already exists, skipping database creation.</li>';
 259                                 } else {
 260                                         if ( $perms < 1 ) {
 261                                                 print '<li>ERROR: the user "' . htmlspecialchars( $superuser ) . '" cannot create databases. ';
 262                                                 print 'Please use a different Postgres user.</li>';
 263                                                 dieout( );
 264                                         }
 265                                         print '<li>Creating database <b>' . htmlspecialchars( $wgDBname ) . '</b>...';
 266                                         $safename = $this->addIdentifierQuotes( $wgDBname );
 267                                         $SQL = "CREATE DATABASE $safename OWNER $safeuser ";
 268                                         $this->doQuery( $SQL );
 269                                         print "OK</li>\n";
 270                                         // Hopefully tsearch2 and plpgsql are in template1...
 271                                 }
 272
 273                                 // Reconnect to check out tsearch2 rights for this user
 274                                 print '<li>Connecting to "' . htmlspecialchars( $wgDBname ) . '" as superuser "' .
 275                                         htmlspecialchars( $superuser ) . '" to check rights...';
 276
 277                                 $connectVars = array();
 278                                 if ( $this->mServer != false && $this->mServer != '' ) {
 279                                         $connectVars['host'] = $this->mServer;
 280                                 }
 281                                 if ( $this->mPort != false && $this->mPort != '' ) {
 282                                         $connectVars['port'] = $this->mPort;
 283                                 }
 284                                 $connectVars['dbname'] = $wgDBname;
 285                                 $connectVars['user'] = $superuser;
 286                                 $connectVars['password'] = $password;
 287
 288                                 @$this->mConn = pg_connect( $this->makeConnectionString( $connectVars ) );
 289                                 if ( !$this->mConn ) {
 290                                         print "<b>FAILED TO CONNECT!</b></li>";
 291                                         dieout( );
 292                                 }
 293                                 print "OK</li>\n";
 294                         }
 295
 296                         // Setup the schema for this user if needed
 297                         $result = $this->schemaExists( $wgDBmwschema );
 298                         $safeschema = $this->addIdentifierQuotes( $wgDBmwschema );
 299                         if ( !$result ) {
 300                                 print '<li>Creating schema <b>' . htmlspecialchars( $wgDBmwschema ) . '</b> ...';
 301                                 $result = $this->doQuery( "CREATE SCHEMA $safeschema AUTHORIZATION $safeuser" );
 302                                 if ( !$result ) {
 303                                         print "<b>FAILED</b>.</li>\n";
 304                                         dieout( );
 305                                 }
 306                                 print "OK</li>\n";
 307                         } else {
 308                                 print "<li>Schema already exists, explicitly granting rights...\n";
 309                                 $safeschema2 = $this->addQuotes( $wgDBmwschema );
 310                                 $SQL = "SELECT 'GRANT ALL ON '||pg_catalog.quote_ident(relname)||' TO $safeuser;'\n".
 311                                         "FROM pg_catalog.pg_class p, pg_catalog.pg_namespace n\n".
 312                                         "WHERE relnamespace = n.oid AND n.nspname = $safeschema2\n".
 313                                         "AND p.relkind IN ('r','S','v')\n";
 314                                 $SQL .= "UNION\n";
 315                                 $SQL .= "SELECT 'GRANT ALL ON FUNCTION '||pg_catalog.quote_ident(proname)||'('||\n".
 316                                         "pg_catalog.oidvectortypes(p.proargtypes)||') TO $safeuser;'\n".
 317                                         "FROM pg_catalog.pg_proc p, pg_catalog.pg_namespace n\n".
 318                                         "WHERE p.pronamespace = n.oid AND n.nspname = $safeschema2";
 319                                 $res = $this->doQuery( $SQL );
 320                                 if ( !$res ) {
 321                                         print "<b>FAILED</b>. Could not set rights for the user.</li>\n";
 322                                         dieout( );
 323                                 }
 324                                 $this->doQuery( "SET search_path = $safeschema" );
 325                                 $rows = $this->numRows( $res );
 326                                 while ( $rows ) {
 327                                         $rows--;
 328                                         $this->doQuery( pg_fetch_result( $res, $rows, 0 ) );
 329                                 }
 330                                 print "OK</li>";
 331                         }
 332
 333                         // Install plpgsql if needed
 334                         $this->setup_plpgsql();
 335
 336                         return true; // Reconnect as regular user
 337
 338                 } // end superuser
 339
 340                 if ( !defined( 'POSTGRES_SEARCHPATH' ) ) {
 341                         // Install plpgsql if needed
 342                         $this->setup_plpgsql();
 343
 344                         // Does the schema already exist? Who owns it?
 345                         $result = $this->schemaExists( $wgDBmwschema );
 346                         if ( !$result ) {
 347                                 print '<li>Creating schema <b>' . htmlspecialchars( $wgDBmwschema ) . '</b> ...';
 348                                 error_reporting( 0 );
 349                                 $safeschema = $this->addIdentifierQuotes( $wgDBmwschema );
 350                                 $result = $this->doQuery( "CREATE SCHEMA $safeschema" );
 351                                 error_reporting( E_ALL );
 352                                 if ( !$result ) {
 353                                         print '<b>FAILED</b>. The user "' . htmlspecialchars( $wgDBuser ) .
 354                                                 '" must be able to access the schema. '.
 355                                                 'You can try making them the owner of the database, or try creating the schema with a '.
 356                                                 'different user, and then grant access to the "' .
 357                                                 htmlspecialchars( $wgDBuser ) . "\" user.</li>\n";
 358                                         dieout( );
 359                                 }
 360                                 print "OK</li>\n";
 361                         } elseif ( $result != $wgDBuser ) {
 362                                 print '<li>Schema "' . htmlspecialchars( $wgDBmwschema ) . '" exists but is not owned by "' .
 363                                         htmlspecialchars( $wgDBuser ) . "\". Not ideal.</li>\n";
 364                         } else {
 365                                 print '<li>Schema "' . htmlspecialchars( $wgDBmwschema ) . '" exists and is owned by "' .
 366                                         htmlspecialchars( $wgDBuser ) . "\". Excellent.</li>\n";
 367                         }
 368
 369                         // Always return GMT time to accomodate the existing integer-based timestamp assumption
 370                         print "<li>Setting the timezone to GMT for user \"" . htmlspecialchars( $wgDBuser ) . '" ...';
 371                         $SQL = "ALTER USER $safeuser SET timezone = 'GMT'";
 372                         $result = pg_query( $this->mConn, $SQL );
 373                         if ( !$result ) {
 374                                 print "<b>FAILED</b>.</li>\n";
 375                                 dieout( );
 376                         }
 377                         print "OK</li>\n";
 378                         // Set for the rest of this session
 379                         $SQL = "SET timezone = 'GMT'";
 380                         $result = pg_query( $this->mConn, $SQL );
 381                         if ( !$result ) {
 382                                 print "<li>Failed to set timezone</li>\n";
 383                                 dieout( );
 384                         }
 385
 386                         print '<li>Setting the datestyle to ISO, YMD for user "' . htmlspecialchars( $wgDBuser ) . '" ...';
 387                         $SQL = "ALTER USER $safeuser SET datestyle = 'ISO, YMD'";
 388                         $result = pg_query( $this->mConn, $SQL );
 389                         if ( !$result ) {
 390                                 print "<b>FAILED</b>.</li>\n";
 391                                 dieout( );
 392                         }
 393                         print "OK</li>\n";
 394                         // Set for the rest of this session
 395                         $SQL = "SET datestyle = 'ISO, YMD'";
 396                         $result = pg_query( $this->mConn, $SQL );
 397                         if ( !$result ) {
 398                                 print "<li>Failed to set datestyle</li>\n";
 399                                 dieout( );
 400                         }
 401
 402                         // Fix up the search paths if needed
 403                         print '<li>Setting the search path for user "' . htmlspecialchars( $wgDBuser ) . '" ...';
 404                         $path = $this->addIdentifierQuotes( $wgDBmwschema );
 405                         if ( $wgDBts2schema !== $wgDBmwschema ) {
 406                                 $path .= ', '. $this->addIdentifierQuotes( $wgDBts2schema );
 407                         }
 408                         if ( $wgDBmwschema !== 'public' && $wgDBts2schema !== 'public' ) {
 409                                 $path .= ', public';
 410                         }
 411                         $SQL = "ALTER USER $safeuser SET search_path = $path";
 412                         $result = pg_query( $this->mConn, $SQL );
 413                         if ( !$result ) {
 414                                 print "<b>FAILED</b>.</li>\n";
 415                                 dieout( );
 416                         }
 417                         print "OK</li>\n";
 418                         // Set for the rest of this session
 419                         $SQL = "SET search_path = $path";
 420                         $result = pg_query( $this->mConn, $SQL );
 421                         if ( !$result ) {
 422                                 print "<li>Failed to set search_path</li>\n";
 423                                 dieout( );
 424                         }
 425                         define( 'POSTGRES_SEARCHPATH', $path );
 426                 }
 427         }
 428
 429         /**
 430          * Closes a database connection, if it is open
 431          * Returns success, true if already closed
 432          */
 433         function close() {
 434                 $this->mOpened = false;
 435                 if ( $this->mConn ) {
 436                         return pg_close( $this->mConn );
 437                 } else {
 438                         return true;
 439                 }
 440         }
 441
 442         function doQuery( $sql ) {
 443                 if ( function_exists( 'mb_convert_encoding' ) ) {
 444                         $sql = mb_convert_encoding( $sql, 'UTF-8' );
 445                 }
 446                 $this->mLastResult = pg_query( $this->mConn, $sql );
 447                 $this->mAffectedRows = null; // use pg_affected_rows(mLastResult)
 448                 return $this->mLastResult;
 449         }
 450
 451         function queryIgnore( $sql, $fname = 'DatabasePostgres::queryIgnore' ) {
 452                 return $this->query( $sql, $fname, true );
 453         }
 454
 455         function freeResult( $res ) {
 456                 if ( $res instanceof ResultWrapper ) {
 457                         $res = $res->result;
 458                 }
 459                 if ( !@pg_free_result( $res ) ) {
 460                         throw new DBUnexpectedError( $this, "Unable to free Postgres result\n" );
 461                 }
 462         }
 463
 464         function fetchObject( $res ) {
 465                 if ( $res instanceof ResultWrapper ) {
 466                         $res = $res->result;
 467                 }
 468                 @$row = pg_fetch_object( $res );
 469                 # FIXME: HACK HACK HACK HACK debug
 470
 471                 # TODO:
 472                 # hashar : not sure if the following test really trigger if the object
 473                 #          fetching failed.
 474                 if( pg_last_error( $this->mConn ) ) {
 475                         throw new DBUnexpectedError( $this, 'SQL error: ' . htmlspecialchars( pg_last_error( $this->mConn ) ) );
 476                 }
 477                 return $row;
 478         }
 479
 480         function fetchRow( $res ) {
 481                 if ( $res instanceof ResultWrapper ) {
 482                         $res = $res->result;
 483                 }
 484                 @$row = pg_fetch_array( $res );
 485                 if( pg_last_error( $this->mConn ) ) {
 486                         throw new DBUnexpectedError( $this, 'SQL error: ' . htmlspecialchars( pg_last_error( $this->mConn ) ) );
 487                 }
 488                 return $row;
 489         }
 490
 491         function numRows( $res ) {
 492                 if ( $res instanceof ResultWrapper ) {
 493                         $res = $res->result;
 494                 }
 495                 @$n = pg_num_rows( $res );
 496                 if( pg_last_error( $this->mConn ) ) {
 497                         throw new DBUnexpectedError( $this, 'SQL error: ' . htmlspecialchars( pg_last_error( $this->mConn ) ) );
 498                 }
 499                 return $n;
 500         }
 501
 502         function numFields( $res ) {
 503                 if ( $res instanceof ResultWrapper ) {
 504                         $res = $res->result;
 505                 }
 506                 return pg_num_fields( $res );
 507         }
 508
 509         function fieldName( $res, $n ) {
 510                 if ( $res instanceof ResultWrapper ) {
 511                         $res = $res->result;
 512                 }
 513                 return pg_field_name( $res, $n );
 514         }
 515
 516         /**
 517          * This must be called after nextSequenceVal
 518          */
 519         function insertId() {
 520                 return $this->mInsertId;
 521         }
 522
 523         function dataSeek( $res, $row ) {
 524                 if ( $res instanceof ResultWrapper ) {
 525                         $res = $res->result;
 526                 }
 527                 return pg_result_seek( $res, $row );
 528         }
 529
 530         function lastError() {
 531                 if ( $this->mConn ) {
 532                         return pg_last_error();
 533                 } else {
 534                         return 'No database connection';
 535                 }
 536         }
 537         function lastErrno() {
 538                 return pg_last_error() ? 1 : 0;
 539         }
 540
 541         function affectedRows() {
 542                 if ( !is_null( $this->mAffectedRows ) ) {
 543                         // Forced result for simulated queries
 544                         return $this->mAffectedRows;
 545                 }
 546                 if( empty( $this->mLastResult ) ) {
 547                         return 0;
 548                 }
 549                 return pg_affected_rows( $this->mLastResult );
 550         }
 551
 552         /**
 553          * Estimate rows in dataset
 554          * Returns estimated count, based on EXPLAIN output
 555          * This is not necessarily an accurate estimate, so use sparingly
 556          * Returns -1 if count cannot be found
 557          * Takes same arguments as Database::select()
 558          */
 559         function estimateRowCount( $table, $vars = '*', $conds='', $fname = 'DatabasePostgres::estimateRowCount', $options = array() ) {
 560                 $options['EXPLAIN'] = true;
 561                 $res = $this->select( $table, $vars, $conds, $fname, $options );
 562                 $rows = -1;
 563                 if ( $res ) {
 564                         $row = $this->fetchRow( $res );
 565                         $count = array();
 566                         if( preg_match( '/rows=(\d+)/', $row[0], $count ) ) {
 567                                 $rows = $count[1];
 568                         }
 569                 }
 570                 return $rows;
 571         }
 572
 573         /**
 574          * Returns information about an index
 575          * If errors are explicitly ignored, returns NULL on failure
 576          */
 577         function indexInfo( $table, $index, $fname = 'DatabasePostgres::indexInfo' ) {
 578                 $sql = "SELECT indexname FROM pg_indexes WHERE tablename='$table'";
 579                 $res = $this->query( $sql, $fname );
 580                 if ( !$res ) {
 581                         return null;
 582                 }
 583                 foreach ( $res as $row ) {
 584                         if ( $row->indexname == $this->indexName( $index ) ) {
 585                                 return $row;
 586                         }
 587                 }
 588                 return false;
 589         }
 590
 591         function indexUnique( $table, $index, $fname = 'DatabasePostgres::indexUnique' ) {
 592                 $sql = "SELECT indexname FROM pg_indexes WHERE tablename='{$table}'".
 593                         " AND indexdef LIKE 'CREATE UNIQUE%(" .
 594                         $this->strencode( $this->indexName( $index ) ) .
 595                         ")'";
 596                 $res = $this->query( $sql, $fname );
 597                 if ( !$res ) {
 598                         return null;
 599                 }
 600                 foreach ( $res as $row ) {
 601                         return true;
 602                 }
 603                 return false;
 604         }
 605
 606         /**
 607          * INSERT wrapper, inserts an array into a table
 608          *
 609          * $args may be a single associative array, or an array of these with numeric keys,
 610          * for multi-row insert (Postgres version 8.2 and above only).
 611          *
 612          * @param $table   String: Name of the table to insert to.
 613          * @param $args    Array: Items to insert into the table.
 614          * @param $fname   String: Name of the function, for profiling
 615          * @param $options String or Array. Valid options: IGNORE
 616          *
 617          * @return bool Success of insert operation. IGNORE always returns true.
 618          */
 619         function insert( $table, $args, $fname = 'DatabasePostgres::insert', $options = array() ) {
 620                 if ( !count( $args ) ) {
 621                         return true;
 622                 }
 623
 624                 $table = $this->tableName( $table );
 625                 if (! isset( $this->numeric_version ) ) {
 626                         $this->getServerVersion();
 627                 }
 628
 629                 if ( !is_array( $options ) ) {
 630                         $options = array( $options );
 631                 }
 632
 633                 if ( isset( $args[0] ) && is_array( $args[0] ) ) {
 634                         $multi = true;
 635                         $keys = array_keys( $args[0] );
 636                 } else {
 637                         $multi = false;
 638                         $keys = array_keys( $args );
 639                 }
 640
 641                 // If IGNORE is set, we use savepoints to emulate mysql's behavior
 642                 $ignore = in_array( 'IGNORE', $options ) ? 'mw' : '';
 643
 644                 // If we are not in a transaction, we need to be for savepoint trickery
 645                 $didbegin = 0;
 646                 if ( $ignore ) {
 647                         if ( !$this->mTrxLevel ) {
 648                                 $this->begin();
 649                                 $didbegin = 1;
 650                         }
 651                         $olde = error_reporting( 0 );
 652                         // For future use, we may want to track the number of actual inserts
 653                         // Right now, insert (all writes) simply return true/false
 654                         $numrowsinserted = 0;
 655                 }
 656
 657                 $sql = "INSERT INTO $table (" . implode( ',', $keys ) . ') VALUES ';
 658
 659                 if ( $multi ) {
 660                         if ( $this->numeric_version >= 8.2 && !$ignore ) {
 661                                 $first = true;
 662                                 foreach ( $args as $row ) {
 663                                         if ( $first ) {
 664                                                 $first = false;
 665                                         } else {
 666                                                 $sql .= ',';
 667                                         }
 668                                         $sql .= '(' . $this->makeList( $row ) . ')';
 669                                 }
 670                                 $res = (bool)$this->query( $sql, $fname, $ignore );
 671                         } else {
 672                                 $res = true;
 673                                 $origsql = $sql;
 674                                 foreach ( $args as $row ) {
 675                                         $tempsql = $origsql;
 676                                         $tempsql .= '(' . $this->makeList( $row ) . ')';
 677
 678                                         if ( $ignore ) {
 679                                                 pg_query( $this->mConn, "SAVEPOINT $ignore" );
 680                                         }
 681
 682                                         $tempres = (bool)$this->query( $tempsql, $fname, $ignore );
 683
 684                                         if ( $ignore ) {
 685                                                 $bar = pg_last_error();
 686                                                 if ( $bar != false ) {
 687                                                         pg_query( $this->mConn, "ROLLBACK TO $ignore" );
 688                                                 } else {
 689                                                         pg_query( $this->mConn, "RELEASE $ignore" );
 690                                                         $numrowsinserted++;
 691                                                 }
 692                                         }
 693
 694                                         // If any of them fail, we fail overall for this function call
 695                                         // Note that this will be ignored if IGNORE is set
 696                                         if ( !$tempres ) {
 697                                                 $res = false;
 698                                         }
 699                                 }
 700                         }
 701                 } else {
 702                         // Not multi, just a lone insert
 703                         if ( $ignore ) {
 704                                 pg_query($this->mConn, "SAVEPOINT $ignore");
 705                         }
 706
 707                         $sql .= '(' . $this->makeList( $args ) . ')';
 708                         $res = (bool)$this->query( $sql, $fname, $ignore );
 709                         if ( $ignore ) {
 710                                 $bar = pg_last_error();
 711                                 if ( $bar != false ) {
 712                                         pg_query( $this->mConn, "ROLLBACK TO $ignore" );
 713                                 } else {
 714                                         pg_query( $this->mConn, "RELEASE $ignore" );
 715                                         $numrowsinserted++;
 716                                 }
 717                         }
 718                 }
 719                 if ( $ignore ) {
 720                         $olde = error_reporting( $olde );
 721                         if ( $didbegin ) {
 722                                 $this->commit();
 723                         }
 724
 725                         // Set the affected row count for the whole operation
 726                         $this->mAffectedRows = $numrowsinserted;
 727
 728                         // IGNORE always returns true
 729                         return true;
 730                 }
 731
 732                 return $res;
 733         }
 734
 735         /**
 736          * INSERT SELECT wrapper
 737          * $varMap must be an associative array of the form array( 'dest1' => 'source1', ...)
 738          * Source items may be literals rather then field names, but strings should be quoted with Database::addQuotes()
 739          * $conds may be "*" to copy the whole table
 740          * srcTable may be an array of tables.
 741          * @todo FIXME: implement this a little better (seperate select/insert)?
 742          */
 743         function insertSelect( $destTable, $srcTable, $varMap, $conds, $fname = 'DatabasePostgres::insertSelect',
 744                 $insertOptions = array(), $selectOptions = array() )
 745         {
 746                 $destTable = $this->tableName( $destTable );
 747
 748                 // If IGNORE is set, we use savepoints to emulate mysql's behavior
 749                 $ignore = in_array( 'IGNORE', $insertOptions ) ? 'mw' : '';
 750
 751                 if( is_array( $insertOptions ) ) {
 752                         $insertOptions = implode( ' ', $insertOptions );
 753                 }
 754                 if( !is_array( $selectOptions ) ) {
 755                         $selectOptions = array( $selectOptions );
 756                 }
 757                 list( $startOpts, $useIndex, $tailOpts ) = $this->makeSelectOptions( $selectOptions );
 758                 if( is_array( $srcTable ) ) {
 759                         $srcTable = implode( ',', array_map( array( &$this, 'tableName' ), $srcTable ) );
 760                 } else {
 761                         $srcTable = $this->tableName( $srcTable );
 762                 }
 763
 764                 // If we are not in a transaction, we need to be for savepoint trickery
 765                 $didbegin = 0;
 766                 if ( $ignore ) {
 767                         if( !$this->mTrxLevel ) {
 768                                 $this->begin();
 769                                 $didbegin = 1;
 770                         }
 771                         $olde = error_reporting( 0 );
 772                         $numrowsinserted = 0;
 773                         pg_query( $this->mConn, "SAVEPOINT $ignore");
 774                 }
 775
 776                 $sql = "INSERT INTO $destTable (" . implode( ',', array_keys( $varMap ) ) . ')' .
 777                                 " SELECT $startOpts " . implode( ',', $varMap ) .
 778                                 " FROM $srcTable $useIndex";
 779
 780                 if ( $conds != '*' ) {
 781                         $sql .= ' WHERE ' . $this->makeList( $conds, LIST_AND );
 782                 }
 783
 784                 $sql .= " $tailOpts";
 785
 786                 $res = (bool)$this->query( $sql, $fname, $ignore );
 787                 if( $ignore ) {
 788                         $bar = pg_last_error();
 789                         if( $bar != false ) {
 790                                 pg_query( $this->mConn, "ROLLBACK TO $ignore" );
 791                         } else {
 792                                 pg_query( $this->mConn, "RELEASE $ignore" );
 793                                 $numrowsinserted++;
 794                         }
 795                         $olde = error_reporting( $olde );
 796                         if( $didbegin ) {
 797                                 $this->commit();
 798                         }
 799
 800                         // Set the affected row count for the whole operation
 801                         $this->mAffectedRows = $numrowsinserted;
 802
 803                         // IGNORE always returns true
 804                         return true;
 805                 }
 806
 807                 return $res;
 808         }
 809
 810         function tableName( $name ) {
 811                 # Replace reserved words with better ones
 812                 switch( $name ) {
 813                         case 'user':
 814                                 return 'mwuser';
 815                         case 'text':
 816                                 return 'pagecontent';
 817                         default:
 818                                 return $name;
 819                 }
 820         }
 821
 822         /**
 823          * Return the next in a sequence, save the value for retrieval via insertId()
 824          */
 825         function nextSequenceValue( $seqName ) {
 826                 $safeseq = str_replace( "'", "''", $seqName );
 827                 $res = $this->query( "SELECT nextval('$safeseq')" );
 828                 $row = $this->fetchRow( $res );
 829                 $this->mInsertId = $row[0];
 830                 return $this->mInsertId;
 831         }
 832
 833         /**
 834          * Return the current value of a sequence. Assumes it has been nextval'ed in this session.
 835          */
 836         function currentSequenceValue( $seqName ) {
 837                 $safeseq = str_replace( "'", "''", $seqName );
 838                 $res = $this->query( "SELECT currval('$safeseq')" );
 839                 $row = $this->fetchRow( $res );
 840                 $currval = $row[0];
 841                 return $currval;
 842         }
 843
 844         /**
 845          * REPLACE query wrapper
 846          * Postgres simulates this with a DELETE followed by INSERT
 847          * $row is the row to insert, an associative array
 848          * $uniqueIndexes is an array of indexes. Each element may be either a
 849          * field name or an array of field names
 850          *
 851          * It may be more efficient to leave off unique indexes which are unlikely to collide.
 852          * However if you do this, you run the risk of encountering errors which wouldn't have
 853          * occurred in MySQL
 854          */
 855         function replace( $table, $uniqueIndexes, $rows, $fname = 'DatabasePostgres::replace' ) {
 856                 $table = $this->tableName( $table );
 857
 858                 if ( count( $rows ) == 0 ) {
 859                         return;
 860                 }
 861
 862                 # Single row case
 863                 if ( !is_array( reset( $rows ) ) ) {
 864                         $rows = array( $rows );
 865                 }
 866
 867                 foreach( $rows as $row ) {
 868                         # Delete rows which collide
 869                         if ( $uniqueIndexes ) {
 870                                 $sql = "DELETE FROM $table WHERE ";
 871                                 $first = true;
 872                                 foreach ( $uniqueIndexes as $index ) {
 873                                         if ( $first ) {
 874                                                 $first = false;
 875                                                 $sql .= '(';
 876                                         } else {
 877                                                 $sql .= ') OR (';
 878                                         }
 879                                         if ( is_array( $index ) ) {
 880                                                 $first2 = true;
 881                                                 foreach ( $index as $col ) {
 882                                                         if ( $first2 ) {
 883                                                                 $first2 = false;
 884                                                         } else {
 885                                                                 $sql .= ' AND ';
 886                                                         }
 887                                                         $sql .= $col.'=' . $this->addQuotes( $row[$col] );
 888                                                 }
 889                                         } else {
 890                                                 $sql .= $index.'=' . $this->addQuotes( $row[$index] );
 891                                         }
 892                                 }
 893                                 $sql .= ')';
 894                                 $this->query( $sql, $fname );
 895                         }
 896
 897                         # Now insert the row
 898                         $sql = "INSERT INTO $table (" . $this->makeList( array_keys( $row ), LIST_NAMES ) .') VALUES (' .
 899                                 $this->makeList( $row, LIST_COMMA ) . ')';
 900                         $this->query( $sql, $fname );
 901                 }
 902         }
 903
 904         # DELETE where the condition is a join
 905         function deleteJoin( $delTable, $joinTable, $delVar, $joinVar, $conds, $fname = 'DatabasePostgres::deleteJoin' ) {
 906                 if ( !$conds ) {
 907                         throw new DBUnexpectedError( $this, 'DatabasePostgres::deleteJoin() called with empty $conds' );
 908                 }
 909
 910                 $delTable = $this->tableName( $delTable );
 911                 $joinTable = $this->tableName( $joinTable );
 912                 $sql = "DELETE FROM $delTable WHERE $delVar IN (SELECT $joinVar FROM $joinTable ";
 913                 if ( $conds != '*' ) {
 914                         $sql .= 'WHERE ' . $this->makeList( $conds, LIST_AND );
 915                 }
 916                 $sql .= ')';
 917
 918                 $this->query( $sql, $fname );
 919         }
 920
 921         # Returns the size of a text field, or -1 for "unlimited"
 922         function textFieldSize( $table, $field ) {
 923                 $table = $this->tableName( $table );
 924                 $sql = "SELECT t.typname as ftype,a.atttypmod as size
 925                         FROM pg_class c, pg_attribute a, pg_type t
 926                         WHERE relname='$table' AND a.attrelid=c.oid AND
 927                                 a.atttypid=t.oid and a.attname='$field'";
 928                 $res =$this->query( $sql );
 929                 $row = $this->fetchObject( $res );
 930                 if ( $row->ftype == 'varchar' ) {
 931                         $size = $row->size - 4;
 932                 } else {
 933                         $size = $row->size;
 934                 }
 935                 return $size;
 936         }
 937
 938         function limitResult( $sql, $limit, $offset = false ) {
 939                 return "$sql LIMIT $limit " . ( is_numeric( $offset ) ? " OFFSET {$offset} " : '' );
 940         }
 941
 942         function wasDeadlock() {
 943                 return $this->lastErrno() == '40P01';
 944         }
 945
 946         function duplicateTableStructure( $oldName, $newName, $temporary = false, $fname = 'DatabasePostgres::duplicateTableStructure' ) {
 947                 return $this->query( 'CREATE ' . ( $temporary ? 'TEMPORARY ' : '' ) . " TABLE $newName (LIKE $oldName INCLUDING DEFAULTS)", $fname );
 948         }
 949
 950         function timestamp( $ts = 0 ) {
 951                 return wfTimestamp( TS_POSTGRES, $ts );
 952         }
 953
 954         /**
 955          * Return aggregated value function call
 956          */
 957         function aggregateValue( $valuedata, $valuename = 'value' ) {
 958                 return $valuedata;
 959         }
 960
 961         function reportQueryError( $error, $errno, $sql, $fname, $tempIgnore = false ) {
 962                 // Ignore errors during error handling to avoid infinite recursion
 963                 $ignore = $this->ignoreErrors( true );
 964                 $this->mErrorCount++;
 965
 966                 if ( $ignore || $tempIgnore ) {
 967                         wfDebug( "SQL ERROR (ignored): $error\n" );
 968                         $this->ignoreErrors( $ignore );
 969                 } else {
 970                         $message = "A database error has occurred.  Did you forget to run maintenance/update.php after upgrading?  See: http://www.mediawiki.org/wiki/Manual:Upgrading#Run_the_update_script\n" .
 971                                 "Query: $sql\n" .
 972                                 "Function: $fname\n" .
 973                                 "Error: $errno $error\n";
 974                         throw new DBUnexpectedError( $this, $message );
 975                 }
 976         }
 977
 978         /**
 979          * @return string wikitext of a link to the server software's web site
 980          */
 981         public static function getSoftwareLink() {
 982                 return '[http://www.postgresql.org/ PostgreSQL]';
 983         }
 984
 985         /**
 986          * @return string Version information from the database
 987          */
 988         function getServerVersion() {
 989                 if ( !isset( $this->numeric_version ) ) {
 990                         $versionInfo = pg_version( $this->mConn );
 991                         if ( version_compare( $versionInfo['client'], '7.4.0', 'lt' ) ) {
 992                                 // Old client, abort install
 993                                 $this->numeric_version = '7.3 or earlier';
 994                         } elseif ( isset( $versionInfo['server'] ) ) {
 995                                 // Normal client
 996                                 $this->numeric_version = $versionInfo['server'];
 997                         } else {
 998                                 // Bug 16937: broken pgsql extension from PHP<5.3
 999                                 $this->numeric_version = pg_parameter_status( $this->mConn, 'server_version' );
1000                         }
1001                 }
1002                 return $this->numeric_version;
1003         }
1004
1005         /**
1006          * Query whether a given relation exists (in the given schema, or the
1007          * default mw one if not given)
1008          */
1009         function relationExists( $table, $types, $schema = false ) {
1010                 global $wgDBmwschema;
1011                 if ( !is_array( $types ) ) {
1012                         $types = array( $types );
1013                 }
1014                 if ( !$schema ) {
1015                         $schema = $wgDBmwschema;
1016                 }
1017                 $etable = $this->addQuotes( $table );
1018                 $eschema = $this->addQuotes( $schema );
1019                 $SQL = "SELECT 1 FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n "
1020                         . "WHERE c.relnamespace = n.oid AND c.relname = $etable AND n.nspname = $eschema "
1021                         . "AND c.relkind IN ('" . implode( "','", $types ) . "')";
1022                 $res = $this->query( $SQL );
1023                 $count = $res ? $res->numRows() : 0;
1024                 return (bool)$count;
1025         }
1026
1027         /**
1028          * For backward compatibility, this function checks both tables and
1029          * views.
1030          */
1031         function tableExists( $table, $schema = false ) {
1032                 return $this->relationExists( $table, array( 'r', 'v' ), $schema );
1033         }
1034
1035         function sequenceExists( $sequence, $schema = false ) {
1036                 return $this->relationExists( $sequence, 'S', $schema );
1037         }
1038
1039         function triggerExists( $table, $trigger ) {
1040                 global $wgDBmwschema;
1041
1042                 $q = <<<SQL
1043         SELECT 1 FROM pg_class, pg_namespace, pg_trigger
1044                 WHERE relnamespace=pg_namespace.oid AND relkind='r'
1045                       AND tgrelid=pg_class.oid
1046                       AND nspname=%s AND relname=%s AND tgname=%s
1047 SQL;
1048                 $res = $this->query(
1049                         sprintf(
1050                                 $q,
1051                                 $this->addQuotes( $wgDBmwschema ),
1052                                 $this->addQuotes( $table ),
1053                                 $this->addQuotes( $trigger )
1054                         )
1055                 );
1056                 if ( !$res ) {
1057                         return null;
1058                 }
1059                 $rows = $res->numRows();
1060                 return $rows;
1061         }
1062
1063         function ruleExists( $table, $rule ) {
1064                 global $wgDBmwschema;
1065                 $exists = $this->selectField( 'pg_rules', 'rulename',
1066                         array(
1067                                 'rulename' => $rule,
1068                                 'tablename' => $table,
1069                                 'schemaname' => $wgDBmwschema
1070                         )
1071                 );
1072                 return $exists === $rule;
1073         }
1074
1075         function constraintExists( $table, $constraint ) {
1076                 global $wgDBmwschema;
1077                 $SQL = sprintf( "SELECT 1 FROM information_schema.table_constraints ".
1078                            "WHERE constraint_schema = %s AND table_name = %s AND constraint_name = %s",
1079                         $this->addQuotes( $wgDBmwschema ),
1080                         $this->addQuotes( $table ),
1081                         $this->addQuotes( $constraint )
1082                 );
1083                 $res = $this->query( $SQL );
1084                 if ( !$res ) {
1085                         return null;
1086                 }
1087                 $rows = $res->numRows();
1088                 return $rows;
1089         }
1090
1091         /**
1092          * Query whether a given schema exists. Returns the name of the owner
1093          */
1094         function schemaExists( $schema ) {
1095                 $eschema = str_replace( "'", "''", $schema );
1096                 $SQL = "SELECT rolname FROM pg_catalog.pg_namespace n, pg_catalog.pg_roles r "
1097                                 ."WHERE n.nspowner=r.oid AND n.nspname = '$eschema'";
1098                 $res = $this->query( $SQL );
1099                 if ( $res && $res->numRows() ) {
1100                         $row = $res->fetchObject();
1101                         $owner = $row->rolname;
1102                 } else {
1103                         $owner = false;
1104                 }
1105                 return $owner;
1106         }
1107
1108         function fieldInfo( $table, $field ) {
1109                 return PostgresField::fromText( $this, $table, $field );
1110         }
1111
1112         /**
1113          * pg_field_type() wrapper
1114          */
1115         function fieldType( $res, $index ) {
1116                 if ( $res instanceof ResultWrapper ) {
1117                         $res = $res->result;
1118                 }
1119                 return pg_field_type( $res, $index );
1120         }
1121
1122         /* Not even sure why this is used in the main codebase... */
1123         function limitResultForUpdate( $sql, $num ) {
1124                 return $sql;
1125         }
1126
1127         function encodeBlob( $b ) {
1128                 return new Blob( pg_escape_bytea( $this->mConn, $b ) );
1129         }
1130
1131         function decodeBlob( $b ) {
1132                 if ( $b instanceof Blob ) {
1133                         $b = $b->fetch();
1134                 }
1135                 return pg_unescape_bytea( $b );
1136         }
1137
1138         function strencode( $s ) { # Should not be called by us
1139                 return pg_escape_string( $this->mConn, $s );
1140         }
1141
1142         function addQuotes( $s ) {
1143                 if ( is_null( $s ) ) {
1144                         return 'NULL';
1145                 } elseif ( is_bool( $s ) ) {
1146                         return intval( $s );
1147                 } elseif ( $s instanceof Blob ) {
1148                         return "'" . $s->fetch( $s ) . "'";
1149                 }
1150                 return "'" . pg_escape_string( $this->mConn, $s ) . "'";
1151         }
1152
1153         /**
1154          * Postgres specific version of replaceVars.
1155          * Calls the parent version in Database.php
1156          *
1157          * @private
1158          *
1159          * @param $ins String: SQL string, read from a stream (usually tables.sql)
1160          *
1161          * @return string SQL string
1162          */
1163         protected function replaceVars( $ins ) {
1164                 $ins = parent::replaceVars( $ins );
1165
1166                 if ( $this->numeric_version >= 8.3 ) {
1167                         // Thanks for not providing backwards-compatibility, 8.3
1168                         $ins = preg_replace( "/to_tsvector\s*\(\s*'default'\s*,/", 'to_tsvector(', $ins );
1169                 }
1170
1171                 if ( $this->numeric_version <= 8.1 ) { // Our minimum version
1172                         $ins = str_replace( 'USING gin', 'USING gist', $ins );
1173                 }
1174
1175                 return $ins;
1176         }
1177
1178         /**
1179          * Various select options
1180          *
1181          * @private
1182          *
1183          * @param $options Array: an associative array of options to be turned into
1184          *              an SQL query, valid keys are listed in the function.
1185          * @return array
1186          */
1187         function makeSelectOptions( $options ) {
1188                 $preLimitTail = $postLimitTail = '';
1189                 $startOpts = $useIndex = '';
1190
1191                 $noKeyOptions = array();
1192                 foreach ( $options as $key => $option ) {
1193                         if ( is_numeric( $key ) ) {
1194                                 $noKeyOptions[$option] = true;
1195                         }
1196                 }
1197
1198                 if ( isset( $options['GROUP BY'] ) ) {
1199                         $preLimitTail .= ' GROUP BY ' . $options['GROUP BY'];
1200                 }
1201                 if ( isset( $options['HAVING'] ) ) {
1202                         $preLimitTail .= " HAVING {$options['HAVING']}";
1203                 }
1204                 if ( isset( $options['ORDER BY'] ) ) {
1205                         $preLimitTail .= ' ORDER BY ' . $options['ORDER BY'];
1206                 }
1207
1208                 //if ( isset( $options['LIMIT'] ) ) {
1209                 //      $tailOpts .= $this->limitResult( '', $options['LIMIT'],
1210                 //              isset( $options['OFFSET'] ) ? $options['OFFSET']
1211                 //              : false );
1212                 //}
1213
1214                 if ( isset( $noKeyOptions['FOR UPDATE'] ) ) {
1215                         $postLimitTail .= ' FOR UPDATE';
1216                 }
1217                 if ( isset( $noKeyOptions['LOCK IN SHARE MODE'] ) ) {
1218                         $postLimitTail .= ' LOCK IN SHARE MODE';
1219                 }
1220                 if ( isset( $noKeyOptions['DISTINCT'] ) || isset( $noKeyOptions['DISTINCTROW'] ) ) {
1221                         $startOpts .= 'DISTINCT';
1222                 }
1223
1224                 return array( $startOpts, $useIndex, $preLimitTail, $postLimitTail );
1225         }
1226
1227         function setFakeMaster( $enabled = true ) {}
1228
1229         function getDBname() {
1230                 return $this->mDBname;
1231         }
1232
1233         function getServer() {
1234                 return $this->mServer;
1235         }
1236
1237         function buildConcat( $stringList ) {
1238                 return implode( ' || ', $stringList );
1239         }
1240
1241         public function getSearchEngine() {
1242                 return 'SearchPostgres';
1243         }
1244 } // end DatabasePostgres class