includes/db/DatabasePostgres.php

   1 <?php
   2 /**
   3  * This is the Postgres database abstraction layer.
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published by
   7  * the Free Software Foundation; either version 2 of the License, or
   8  * (at your option) any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13  * GNU General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License along
  16  * with this program; if not, write to the Free Software Foundation, Inc.,
  17  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
  18  * http://www.gnu.org/copyleft/gpl.html
  19  *
  20  * @file
  21  * @ingroup Database
  22  */
  23
  24 class PostgresField implements Field {
  25         private $name, $tablename, $type, $nullable, $max_length, $deferred, $deferrable, $conname,
  26                 $has_default, $default;
  27
  28         /**
  29          * @param $db DatabaseBase
  30          * @param  $table
  31          * @param  $field
  32          * @return null|PostgresField
  33          */
  34         static function fromText( $db, $table, $field ) {
  35                 $q = <<<SQL
  36 SELECT
  37  attnotnull, attlen, conname AS conname,
  38  atthasdef,
  39  adsrc,
  40  COALESCE(condeferred, 'f') AS deferred,
  41  COALESCE(condeferrable, 'f') AS deferrable,
  42  CASE WHEN typname = 'int2' THEN 'smallint'
  43   WHEN typname = 'int4' THEN 'integer'
  44   WHEN typname = 'int8' THEN 'bigint'
  45   WHEN typname = 'bpchar' THEN 'char'
  46  ELSE typname END AS typname
  47 FROM pg_class c
  48 JOIN pg_namespace n ON (n.oid = c.relnamespace)
  49 JOIN pg_attribute a ON (a.attrelid = c.oid)
  50 JOIN pg_type t ON (t.oid = a.atttypid)
  51 LEFT JOIN pg_constraint o ON (o.conrelid = c.oid AND a.attnum = ANY(o.conkey) AND o.contype = 'f')
  52 LEFT JOIN pg_attrdef d on c.oid=d.adrelid and a.attnum=d.adnum
  53 WHERE relkind = 'r'
  54 AND nspname=%s
  55 AND relname=%s
  56 AND attname=%s;
  57 SQL;
  58
  59                 $table = $db->tableName( $table, 'raw' );
  60                 $res = $db->query(
  61                         sprintf( $q,
  62                                 $db->addQuotes( $db->getCoreSchema() ),
  63                                 $db->addQuotes( $table ),
  64                                 $db->addQuotes( $field )
  65                         )
  66                 );
  67                 $row = $db->fetchObject( $res );
  68                 if ( !$row ) {
  69                         return null;
  70                 }
  71                 $n = new PostgresField;
  72                 $n->type = $row->typname;
  73                 $n->nullable = ( $row->attnotnull == 'f' );
  74                 $n->name = $field;
  75                 $n->tablename = $table;
  76                 $n->max_length = $row->attlen;
  77                 $n->deferrable = ( $row->deferrable == 't' );
  78                 $n->deferred = ( $row->deferred == 't' );
  79                 $n->conname = $row->conname;
  80                 $n->has_default = ( $row->atthasdef === 't' );
  81                 $n->default = $row->adsrc;
  82                 return $n;
  83         }
  84
  85         function name() {
  86                 return $this->name;
  87         }
  88
  89         function tableName() {
  90                 return $this->tablename;
  91         }
  92
  93         function type() {
  94                 return $this->type;
  95         }
  96
  97         function isNullable() {
  98                 return $this->nullable;
  99         }
 100
 101         function maxLength() {
 102                 return $this->max_length;
 103         }
 104
 105         function is_deferrable() {
 106                 return $this->deferrable;
 107         }
 108
 109         function is_deferred() {
 110                 return $this->deferred;
 111         }
 112
 113         function conname() {
 114                 return $this->conname;
 115         }
 116         /**
 117          * @since 1.19
 118          */
 119         function defaultValue() {
 120                 if( $this->has_default ) {
 121                         return $this->default;
 122                 } else {
 123                         return false;
 124                 }
 125         }
 126
 127 }
 128
 129 /**
 130  * Used to debug transaction processing
 131  * Only used if $wgDebugDBTransactions is true
 132  *
 133  * @since 1.19
 134  * @ingroup Database
 135  */
 136 class PostgresTransactionState {
 137
 138         static $WATCHED = array(
 139                 array(
 140                         "desc" => "Connection state changed from %s -> %s\n",
 141                         "states" => array(
 142                                 PGSQL_CONNECTION_OK       => "OK",
 143                                 PGSQL_CONNECTION_BAD      => "BAD"
 144                         )
 145                 ),
 146                 array(
 147                         "desc" => "Transaction state changed from %s -> %s\n",
 148                         "states" => array(
 149                                 PGSQL_TRANSACTION_IDLE    => "IDLE",
 150                                 PGSQL_TRANSACTION_ACTIVE  => "ACTIVE",
 151                                 PGSQL_TRANSACTION_INTRANS => "TRANS",
 152                                 PGSQL_TRANSACTION_INERROR => "ERROR",
 153                                 PGSQL_TRANSACTION_UNKNOWN => "UNKNOWN"
 154                         )
 155                 )
 156         );
 157
 158         public function __construct( $conn ) {
 159                 $this->mConn = $conn;
 160                 $this->update();
 161                 $this->mCurrentState = $this->mNewState;
 162         }
 163
 164         public function update() {
 165                 $this->mNewState = array(
 166                         pg_connection_status( $this->mConn ),
 167                         pg_transaction_status( $this->mConn )
 168                 );
 169         }
 170
 171         public function check() {
 172                 global $wgDebugDBTransactions;
 173                 $this->update();
 174                 if ( $wgDebugDBTransactions ) {
 175                         if ( $this->mCurrentState !== $this->mNewState ) {
 176                                 $old = reset( $this->mCurrentState );
 177                                 $new = reset( $this->mNewState );
 178                                 foreach ( self::$WATCHED as $watched ) {
 179                                         if ($old !== $new) {
 180                                                 $this->log_changed($old, $new, $watched);
 181                                         }
 182                                         $old = next( $this->mCurrentState );
 183                                         $new = next( $this->mNewState );
 184
 185                                 }
 186                         }
 187                 }
 188                 $this->mCurrentState = $this->mNewState;
 189         }
 190
 191         protected function describe_changed( $status, $desc_table ) {
 192                 if( isset( $desc_table[$status] ) ) {
 193                         return $desc_table[$status];
 194                 } else {
 195                         return "STATUS " . $status;
 196                 }
 197         }
 198
 199         protected function log_changed( $old, $new, $watched ) {
 200                 wfDebug(sprintf($watched["desc"],
 201                         $this->describe_changed( $old, $watched["states"] ),
 202                         $this->describe_changed( $new, $watched["states"] ))
 203                 );
 204         }
 205 }
 206
 207 /**
 208  * @ingroup Database
 209  */
 210 class DatabasePostgres extends DatabaseBase {
 211         var $mInsertId = null;
 212         var $mLastResult = null;
 213         var $numeric_version = null;
 214         var $mAffectedRows = null;
 215
 216         function getType() {
 217                 return 'postgres';
 218         }
 219
 220         function cascadingDeletes() {
 221                 return true;
 222         }
 223         function cleanupTriggers() {
 224                 return true;
 225         }
 226         function strictIPs() {
 227                 return true;
 228         }
 229         function realTimestamps() {
 230                 return true;
 231         }
 232         function implicitGroupby() {
 233                 return false;
 234         }
 235         function implicitOrderby() {
 236                 return false;
 237         }
 238         function searchableIPs() {
 239                 return true;
 240         }
 241         function functionalIndexes() {
 242                 return true;
 243         }
 244
 245         function hasConstraint( $name ) {
 246                 $SQL = "SELECT 1 FROM pg_catalog.pg_constraint c, pg_catalog.pg_namespace n WHERE c.connamespace = n.oid AND conname = '" .
 247                                 pg_escape_string( $this->mConn, $name ) . "' AND n.nspname = '" . pg_escape_string( $this->mConn, $this->getCoreSchema() ) ."'";
 248                 $res = $this->doQuery( $SQL );
 249                 return $this->numRows( $res );
 250         }
 251
 252         /**
 253          * Usually aborts on failure
 254          * @return DatabaseBase|null
 255          */
 256         function open( $server, $user, $password, $dbName ) {
 257                 # Test for Postgres support, to avoid suppressed fatal error
 258                 if ( !function_exists( 'pg_connect' ) ) {
 259                         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" );
 260                 }
 261
 262                 global $wgDBport;
 263
 264                 if ( !strlen( $user ) ) { # e.g. the class is being loaded
 265                         return;
 266                 }
 267
 268                 $this->mServer = $server;
 269                 $port = $wgDBport;
 270                 $this->mUser = $user;
 271                 $this->mPassword = $password;
 272                 $this->mDBname = $dbName;
 273
 274                 $connectVars = array(
 275                         'dbname' => $dbName,
 276                         'user' => $user,
 277                         'password' => $password
 278                 );
 279                 if ( $server != false && $server != '' ) {
 280                         $connectVars['host'] = $server;
 281                 }
 282                 if ( $port != false && $port != '' ) {
 283                         $connectVars['port'] = $port;
 284                 }
 285                 $this->connectString = $this->makeConnectionString( $connectVars, PGSQL_CONNECT_FORCE_NEW );
 286                 $this->close();
 287                 $this->installErrorHandler();
 288                 $this->mConn = pg_connect( $this->connectString );
 289                 $phpError = $this->restoreErrorHandler();
 290
 291                 if ( !$this->mConn ) {
 292                         wfDebug( "DB connection error\n" );
 293                         wfDebug( "Server: $server, Database: $dbName, User: $user, Password: " . substr( $password, 0, 3 ) . "...\n" );
 294                         wfDebug( $this->lastError() . "\n" );
 295                         throw new DBConnectionError( $this, str_replace( "\n", ' ', $phpError ) );
 296                 }
 297
 298                 $this->mOpened = true;
 299                 $this->mTransactionState = new PostgresTransactionState( $this->mConn );
 300
 301                 global $wgCommandLineMode;
 302                 # If called from the command-line (e.g. importDump), only show errors
 303                 if ( $wgCommandLineMode ) {
 304                         $this->doQuery( "SET client_min_messages = 'ERROR'" );
 305                 }
 306
 307                 $this->query( "SET client_encoding='UTF8'", __METHOD__ );
 308                 $this->query( "SET datestyle = 'ISO, YMD'", __METHOD__ );
 309                 $this->query( "SET timezone = 'GMT'", __METHOD__ );
 310                 $this->query( "SET standard_conforming_strings = on", __METHOD__ );
 311
 312                 global $wgDBmwschema;
 313                 $this->determineCoreSchema( $wgDBmwschema );
 314
 315                 return $this->mConn;
 316         }
 317
 318         /**
 319          * Postgres doesn't support selectDB in the same way MySQL does. So if the
 320          * DB name doesn't match the open connection, open a new one
 321          * @return
 322          */
 323         function selectDB( $db ) {
 324                 if ( $this->mDBname !== $db ) {
 325                         return (bool)$this->open( $this->mServer, $this->mUser, $this->mPassword, $db );
 326                 } else {
 327                         return true;
 328                 }
 329         }
 330
 331         function makeConnectionString( $vars ) {
 332                 $s = '';
 333                 foreach ( $vars as $name => $value ) {
 334                         $s .= "$name='" . str_replace( "'", "\\'", $value ) . "' ";
 335                 }
 336                 return $s;
 337         }
 338
 339         /**
 340          * Closes a database connection, if it is open
 341          * Returns success, true if already closed
 342          * @return bool
 343          */
 344         protected function closeConnection() {
 345                 return pg_close( $this->mConn );
 346         }
 347
 348         protected function doQuery( $sql ) {
 349                 if ( function_exists( 'mb_convert_encoding' ) ) {
 350                         $sql = mb_convert_encoding( $sql, 'UTF-8' );
 351                 }
 352                 $this->mTransactionState->check();
 353                 if( pg_send_query( $this->mConn, $sql ) === false ) {
 354                         throw new DBUnexpectedError( $this, "Unable to post new query to PostgreSQL\n" );
 355                 }
 356                 $this->mLastResult = pg_get_result( $this->mConn );
 357                 $this->mTransactionState->check();
 358                 $this->mAffectedRows = null;
 359                 if ( pg_result_error( $this->mLastResult ) ) {
 360                         return false;
 361                 }
 362                 return $this->mLastResult;
 363         }
 364
 365         protected function dumpError () {
 366                 $diags = array( PGSQL_DIAG_SEVERITY,
 367                                 PGSQL_DIAG_SQLSTATE,
 368                                 PGSQL_DIAG_MESSAGE_PRIMARY,
 369                                 PGSQL_DIAG_MESSAGE_DETAIL,
 370                                 PGSQL_DIAG_MESSAGE_HINT,
 371                                 PGSQL_DIAG_STATEMENT_POSITION,
 372                                 PGSQL_DIAG_INTERNAL_POSITION,
 373                                 PGSQL_DIAG_INTERNAL_QUERY,
 374                                 PGSQL_DIAG_CONTEXT,
 375                                 PGSQL_DIAG_SOURCE_FILE,
 376                                 PGSQL_DIAG_SOURCE_LINE,
 377                                 PGSQL_DIAG_SOURCE_FUNCTION );
 378                 foreach ( $diags as $d ) {
 379                         wfDebug( sprintf("PgSQL ERROR(%d): %s\n", $d, pg_result_error_field( $this->mLastResult, $d ) ) );
 380                 }
 381         }
 382
 383         function reportQueryError( $error, $errno, $sql, $fname, $tempIgnore = false ) {
 384                 /* Transaction stays in the ERROR state until rolledback */
 385                 if ( $tempIgnore ) {
 386                         /* Check for constraint violation */
 387                         if ( $errno === '23505' ) {
 388                                 parent::reportQueryError( $error, $errno, $sql, $fname, $tempIgnore );
 389                                 return;
 390                         }
 391                 }
 392                 /* Don't ignore serious errors */
 393                 $this->rollback( __METHOD__ );
 394                 parent::reportQueryError( $error, $errno, $sql, $fname, false );
 395         }
 396
 397
 398         function queryIgnore( $sql, $fname = 'DatabasePostgres::queryIgnore' ) {
 399                 return $this->query( $sql, $fname, true );
 400         }
 401
 402         function freeResult( $res ) {
 403                 if ( $res instanceof ResultWrapper ) {
 404                         $res = $res->result;
 405                 }
 406                 wfSuppressWarnings();
 407                 $ok = pg_free_result( $res );
 408                 wfRestoreWarnings();
 409                 if ( !$ok ) {
 410                         throw new DBUnexpectedError( $this, "Unable to free Postgres result\n" );
 411                 }
 412         }
 413
 414         function fetchObject( $res ) {
 415                 if ( $res instanceof ResultWrapper ) {
 416                         $res = $res->result;
 417                 }
 418                 wfSuppressWarnings();
 419                 $row = pg_fetch_object( $res );
 420                 wfRestoreWarnings();
 421                 # @todo FIXME: HACK HACK HACK HACK debug
 422
 423                 # @todo hashar: not sure if the following test really trigger if the object
 424                 #          fetching failed.
 425                 if( pg_last_error( $this->mConn ) ) {
 426                         throw new DBUnexpectedError( $this, 'SQL error: ' . htmlspecialchars( pg_last_error( $this->mConn ) ) );
 427                 }
 428                 return $row;
 429         }
 430
 431         function fetchRow( $res ) {
 432                 if ( $res instanceof ResultWrapper ) {
 433                         $res = $res->result;
 434                 }
 435                 wfSuppressWarnings();
 436                 $row = pg_fetch_array( $res );
 437                 wfRestoreWarnings();
 438                 if( pg_last_error( $this->mConn ) ) {
 439                         throw new DBUnexpectedError( $this, 'SQL error: ' . htmlspecialchars( pg_last_error( $this->mConn ) ) );
 440                 }
 441                 return $row;
 442         }
 443
 444         function numRows( $res ) {
 445                 if ( $res instanceof ResultWrapper ) {
 446                         $res = $res->result;
 447                 }
 448                 wfSuppressWarnings();
 449                 $n = pg_num_rows( $res );
 450                 wfRestoreWarnings();
 451                 if( pg_last_error( $this->mConn ) ) {
 452                         throw new DBUnexpectedError( $this, 'SQL error: ' . htmlspecialchars( pg_last_error( $this->mConn ) ) );
 453                 }
 454                 return $n;
 455         }
 456
 457         function numFields( $res ) {
 458                 if ( $res instanceof ResultWrapper ) {
 459                         $res = $res->result;
 460                 }
 461                 return pg_num_fields( $res );
 462         }
 463
 464         function fieldName( $res, $n ) {
 465                 if ( $res instanceof ResultWrapper ) {
 466                         $res = $res->result;
 467                 }
 468                 return pg_field_name( $res, $n );
 469         }
 470
 471         /**
 472          * This must be called after nextSequenceVal
 473          * @return null
 474          */
 475         function insertId() {
 476                 return $this->mInsertId;
 477         }
 478
 479         function dataSeek( $res, $row ) {
 480                 if ( $res instanceof ResultWrapper ) {
 481                         $res = $res->result;
 482                 }
 483                 return pg_result_seek( $res, $row );
 484         }
 485
 486         function lastError() {
 487                 if ( $this->mConn ) {
 488                         if ( $this->mLastResult ) {
 489                                 return pg_result_error( $this->mLastResult );
 490                         } else {
 491                                 return pg_last_error();
 492                         }
 493                 } else {
 494                         return 'No database connection';
 495                 }
 496         }
 497         function lastErrno() {
 498                 if ( $this->mLastResult ) {
 499                         return pg_result_error_field( $this->mLastResult, PGSQL_DIAG_SQLSTATE );
 500                 } else {
 501                         return false;
 502                 }
 503         }
 504
 505         function affectedRows() {
 506                 if ( !is_null( $this->mAffectedRows ) ) {
 507                         // Forced result for simulated queries
 508                         return $this->mAffectedRows;
 509                 }
 510                 if( empty( $this->mLastResult ) ) {
 511                         return 0;
 512                 }
 513                 return pg_affected_rows( $this->mLastResult );
 514         }
 515
 516         /**
 517          * Estimate rows in dataset
 518          * Returns estimated count, based on EXPLAIN output
 519          * This is not necessarily an accurate estimate, so use sparingly
 520          * Returns -1 if count cannot be found
 521          * Takes same arguments as Database::select()
 522          * @return int
 523          */
 524         function estimateRowCount( $table, $vars = '*', $conds='', $fname = 'DatabasePostgres::estimateRowCount', $options = array() ) {
 525                 $options['EXPLAIN'] = true;
 526                 $res = $this->select( $table, $vars, $conds, $fname, $options );
 527                 $rows = -1;
 528                 if ( $res ) {
 529                         $row = $this->fetchRow( $res );
 530                         $count = array();
 531                         if( preg_match( '/rows=(\d+)/', $row[0], $count ) ) {
 532                                 $rows = $count[1];
 533                         }
 534                 }
 535                 return $rows;
 536         }
 537
 538         /**
 539          * Returns information about an index
 540          * If errors are explicitly ignored, returns NULL on failure
 541          * @return bool|null
 542          */
 543         function indexInfo( $table, $index, $fname = 'DatabasePostgres::indexInfo' ) {
 544                 $sql = "SELECT indexname FROM pg_indexes WHERE tablename='$table'";
 545                 $res = $this->query( $sql, $fname );
 546                 if ( !$res ) {
 547                         return null;
 548                 }
 549                 foreach ( $res as $row ) {
 550                         if ( $row->indexname == $this->indexName( $index ) ) {
 551                                 return $row;
 552                         }
 553                 }
 554                 return false;
 555         }
 556
 557         /**
 558          * Returns is of attributes used in index
 559          *
 560          * @since 1.19
 561          * @return Array
 562          */
 563         function indexAttributes ( $index, $schema = false ) {
 564                 if ( $schema === false )
 565                         $schema = $this->getCoreSchema();
 566                 /*
 567                  * A subquery would be not needed if we didn't care about the order
 568                  * of attributes, but we do
 569                  */
 570                 $sql = <<<__INDEXATTR__
 571
 572                         SELECT opcname,
 573                                 attname,
 574                                 i.indoption[s.g] as option,
 575                                 pg_am.amname
 576                         FROM
 577                                 (SELECT generate_series(array_lower(isub.indkey,1), array_upper(isub.indkey,1)) AS g
 578                                         FROM
 579                                                 pg_index isub
 580                                         JOIN pg_class cis
 581                                                 ON cis.oid=isub.indexrelid
 582                                         JOIN pg_namespace ns
 583                                                 ON cis.relnamespace = ns.oid
 584                                         WHERE cis.relname='$index' AND ns.nspname='$schema') AS s,
 585                                 pg_attribute,
 586                                 pg_opclass opcls,
 587                                 pg_am,
 588                                 pg_class ci
 589                                 JOIN pg_index i
 590                                         ON ci.oid=i.indexrelid
 591                                 JOIN pg_class ct
 592                                         ON ct.oid = i.indrelid
 593                                 JOIN pg_namespace n
 594                                         ON ci.relnamespace = n.oid
 595                                 WHERE
 596                                         ci.relname='$index' AND n.nspname='$schema'
 597                                         AND     attrelid = ct.oid
 598                                         AND     i.indkey[s.g] = attnum
 599                                         AND     i.indclass[s.g] = opcls.oid
 600                                         AND     pg_am.oid = opcls.opcmethod
 601 __INDEXATTR__;
 602                 $res = $this->query($sql, __METHOD__);
 603                 $a = array();
 604                 if ( $res ) {
 605                         foreach ( $res as $row ) {
 606                                 $a[] = array(
 607                                         $row->attname,
 608                                         $row->opcname,
 609                                         $row->amname,
 610                                         $row->option);
 611                         }
 612                 } else {
 613                         return null;
 614                 }
 615                 return $a;
 616         }
 617
 618
 619         function indexUnique( $table, $index, $fname = 'DatabasePostgres::indexUnique' ) {
 620                 $sql = "SELECT indexname FROM pg_indexes WHERE tablename='{$table}'".
 621                         " AND indexdef LIKE 'CREATE UNIQUE%(" .
 622                         $this->strencode( $this->indexName( $index ) ) .
 623                         ")'";
 624                 $res = $this->query( $sql, $fname );
 625                 if ( !$res ) {
 626                         return null;
 627                 }
 628                 foreach ( $res as $row ) {
 629                         return true;
 630                 }
 631                 return false;
 632         }
 633
 634         /**
 635          * INSERT wrapper, inserts an array into a table
 636          *
 637          * $args may be a single associative array, or an array of these with numeric keys,
 638          * for multi-row insert (Postgres version 8.2 and above only).
 639          *
 640          * @param $table   String: Name of the table to insert to.
 641          * @param $args    Array: Items to insert into the table.
 642          * @param $fname   String: Name of the function, for profiling
 643          * @param $options String or Array. Valid options: IGNORE
 644          *
 645          * @return bool Success of insert operation. IGNORE always returns true.
 646          */
 647         function insert( $table, $args, $fname = 'DatabasePostgres::insert', $options = array() ) {
 648                 if ( !count( $args ) ) {
 649                         return true;
 650                 }
 651
 652                 $table = $this->tableName( $table );
 653                 if (! isset( $this->numeric_version ) ) {
 654                         $this->getServerVersion();
 655                 }
 656
 657                 if ( !is_array( $options ) ) {
 658                         $options = array( $options );
 659                 }
 660
 661                 if ( isset( $args[0] ) && is_array( $args[0] ) ) {
 662                         $multi = true;
 663                         $keys = array_keys( $args[0] );
 664                 } else {
 665                         $multi = false;
 666                         $keys = array_keys( $args );
 667                 }
 668
 669                 // If IGNORE is set, we use savepoints to emulate mysql's behavior
 670                 $ignore = in_array( 'IGNORE', $options ) ? 'mw' : '';
 671
 672                 // If we are not in a transaction, we need to be for savepoint trickery
 673                 $didbegin = 0;
 674                 if ( $ignore ) {
 675                         if ( !$this->mTrxLevel ) {
 676                                 $this->begin( __METHOD__ );
 677                                 $didbegin = 1;
 678                         }
 679                         $olde = error_reporting( 0 );
 680                         // For future use, we may want to track the number of actual inserts
 681                         // Right now, insert (all writes) simply return true/false
 682                         $numrowsinserted = 0;
 683                 }
 684
 685                 $sql = "INSERT INTO $table (" . implode( ',', $keys ) . ') VALUES ';
 686
 687                 if ( $multi ) {
 688                         if ( $this->numeric_version >= 8.2 && !$ignore ) {
 689                                 $first = true;
 690                                 foreach ( $args as $row ) {
 691                                         if ( $first ) {
 692                                                 $first = false;
 693                                         } else {
 694                                                 $sql .= ',';
 695                                         }
 696                                         $sql .= '(' . $this->makeList( $row ) . ')';
 697                                 }
 698                                 $res = (bool)$this->query( $sql, $fname, $ignore );
 699                         } else {
 700                                 $res = true;
 701                                 $origsql = $sql;
 702                                 foreach ( $args as $row ) {
 703                                         $tempsql = $origsql;
 704                                         $tempsql .= '(' . $this->makeList( $row ) . ')';
 705
 706                                         if ( $ignore ) {
 707                                                 $this->doQuery( "SAVEPOINT $ignore" );
 708                                         }
 709
 710                                         $tempres = (bool)$this->query( $tempsql, $fname, $ignore );
 711
 712                                         if ( $ignore ) {
 713                                                 $bar = pg_last_error();
 714                                                 if ( $bar != false ) {
 715                                                         $this->doQuery( $this->mConn, "ROLLBACK TO $ignore" );
 716                                                 } else {
 717                                                         $this->doQuery( $this->mConn, "RELEASE $ignore" );
 718                                                         $numrowsinserted++;
 719                                                 }
 720                                         }
 721
 722                                         // If any of them fail, we fail overall for this function call
 723                                         // Note that this will be ignored if IGNORE is set
 724                                         if ( !$tempres ) {
 725                                                 $res = false;
 726                                         }
 727                                 }
 728                         }
 729                 } else {
 730                         // Not multi, just a lone insert
 731                         if ( $ignore ) {
 732                                 $this->doQuery( "SAVEPOINT $ignore" );
 733                         }
 734
 735                         $sql .= '(' . $this->makeList( $args ) . ')';
 736                         $res = (bool)$this->query( $sql, $fname, $ignore );
 737                         if ( $ignore ) {
 738                                 $bar = pg_last_error();
 739                                 if ( $bar != false ) {
 740                                         $this->doQuery( "ROLLBACK TO $ignore" );
 741                                 } else {
 742                                         $this->doQuery( "RELEASE $ignore" );
 743                                         $numrowsinserted++;
 744                                 }
 745                         }
 746                 }
 747                 if ( $ignore ) {
 748                         $olde = error_reporting( $olde );
 749                         if ( $didbegin ) {
 750                                 $this->commit( __METHOD__ );
 751                         }
 752
 753                         // Set the affected row count for the whole operation
 754                         $this->mAffectedRows = $numrowsinserted;
 755
 756                         // IGNORE always returns true
 757                         return true;
 758                 }
 759
 760                 return $res;
 761         }
 762
 763         /**
 764          * INSERT SELECT wrapper
 765          * $varMap must be an associative array of the form array( 'dest1' => 'source1', ...)
 766          * Source items may be literals rather then field names, but strings should be quoted with Database::addQuotes()
 767          * $conds may be "*" to copy the whole table
 768          * srcTable may be an array of tables.
 769          * @todo FIXME: Implement this a little better (seperate select/insert)?
 770          * @return bool
 771          */
 772         function insertSelect( $destTable, $srcTable, $varMap, $conds, $fname = 'DatabasePostgres::insertSelect',
 773                 $insertOptions = array(), $selectOptions = array() )
 774         {
 775                 $destTable = $this->tableName( $destTable );
 776
 777                 // If IGNORE is set, we use savepoints to emulate mysql's behavior
 778                 $ignore = in_array( 'IGNORE', $insertOptions ) ? 'mw' : '';
 779
 780                 if( is_array( $insertOptions ) ) {
 781                         $insertOptions = implode( ' ', $insertOptions ); // FIXME: This is unused
 782                 }
 783                 if( !is_array( $selectOptions ) ) {
 784                         $selectOptions = array( $selectOptions );
 785                 }
 786                 list( $startOpts, $useIndex, $tailOpts ) = $this->makeSelectOptions( $selectOptions );
 787                 if( is_array( $srcTable ) ) {
 788                         $srcTable = implode( ',', array_map( array( &$this, 'tableName' ), $srcTable ) );
 789                 } else {
 790                         $srcTable = $this->tableName( $srcTable );
 791                 }
 792
 793                 // If we are not in a transaction, we need to be for savepoint trickery
 794                 $didbegin = 0;
 795                 if ( $ignore ) {
 796                         if( !$this->mTrxLevel ) {
 797                                 $this->begin( __METHOD__ );
 798                                 $didbegin = 1;
 799                         }
 800                         $olde = error_reporting( 0 );
 801                         $numrowsinserted = 0;
 802                         $this->doQuery( "SAVEPOINT $ignore" );
 803                 }
 804
 805                 $sql = "INSERT INTO $destTable (" . implode( ',', array_keys( $varMap ) ) . ')' .
 806                                 " SELECT $startOpts " . implode( ',', $varMap ) .
 807                                 " FROM $srcTable $useIndex";
 808
 809                 if ( $conds != '*' ) {
 810                         $sql .= ' WHERE ' . $this->makeList( $conds, LIST_AND );
 811                 }
 812
 813                 $sql .= " $tailOpts";
 814
 815                 $res = (bool)$this->query( $sql, $fname, $ignore );
 816                 if( $ignore ) {
 817                         $bar = pg_last_error();
 818                         if( $bar != false ) {
 819                                 $this->doQuery( "ROLLBACK TO $ignore" );
 820                         } else {
 821                                 $this->doQuery( "RELEASE $ignore" );
 822                                 $numrowsinserted++;
 823                         }
 824                         $olde = error_reporting( $olde );
 825                         if( $didbegin ) {
 826                                 $this->commit( __METHOD__ );
 827                         }
 828
 829                         // Set the affected row count for the whole operation
 830                         $this->mAffectedRows = $numrowsinserted;
 831
 832                         // IGNORE always returns true
 833                         return true;
 834                 }
 835
 836                 return $res;
 837         }
 838
 839         function tableName( $name, $format = 'quoted' ) {
 840                 # Replace reserved words with better ones
 841                 switch( $name ) {
 842                         case 'user':
 843                                 return $this->realTableName( 'mwuser', $format );
 844                         case 'text':
 845                                 return $this->realTableName( 'pagecontent', $format );
 846                         default:
 847                                 return $this->realTableName( $name, $format );
 848                 }
 849         }
 850
 851         /* Don't cheat on installer */
 852         function realTableName( $name, $format = 'quoted' ) {
 853                 return parent::tableName( $name, $format );
 854         }
 855
 856         /**
 857          * Return the next in a sequence, save the value for retrieval via insertId()
 858          * @return null
 859          */
 860         function nextSequenceValue( $seqName ) {
 861                 $safeseq = str_replace( "'", "''", $seqName );
 862                 $res = $this->query( "SELECT nextval('$safeseq')" );
 863                 $row = $this->fetchRow( $res );
 864                 $this->mInsertId = $row[0];
 865                 return $this->mInsertId;
 866         }
 867
 868         /**
 869          * Return the current value of a sequence. Assumes it has been nextval'ed in this session.
 870          * @return
 871          */
 872         function currentSequenceValue( $seqName ) {
 873                 $safeseq = str_replace( "'", "''", $seqName );
 874                 $res = $this->query( "SELECT currval('$safeseq')" );
 875                 $row = $this->fetchRow( $res );
 876                 $currval = $row[0];
 877                 return $currval;
 878         }
 879
 880         # Returns the size of a text field, or -1 for "unlimited"
 881         function textFieldSize( $table, $field ) {
 882                 $table = $this->tableName( $table );
 883                 $sql = "SELECT t.typname as ftype,a.atttypmod as size
 884                         FROM pg_class c, pg_attribute a, pg_type t
 885                         WHERE relname='$table' AND a.attrelid=c.oid AND
 886                                 a.atttypid=t.oid and a.attname='$field'";
 887                 $res =$this->query( $sql );
 888                 $row = $this->fetchObject( $res );
 889                 if ( $row->ftype == 'varchar' ) {
 890                         $size = $row->size - 4;
 891                 } else {
 892                         $size = $row->size;
 893                 }
 894                 return $size;
 895         }
 896
 897         function limitResult( $sql, $limit, $offset = false ) {
 898                 return "$sql LIMIT $limit " . ( is_numeric( $offset ) ? " OFFSET {$offset} " : '' );
 899         }
 900
 901         function wasDeadlock() {
 902                 return $this->lastErrno() == '40P01';
 903         }
 904
 905         function duplicateTableStructure( $oldName, $newName, $temporary = false, $fname = 'DatabasePostgres::duplicateTableStructure' ) {
 906                 $newName = $this->addIdentifierQuotes( $newName );
 907                 $oldName = $this->addIdentifierQuotes( $oldName );
 908                 return $this->query( 'CREATE ' . ( $temporary ? 'TEMPORARY ' : '' ) . " TABLE $newName (LIKE $oldName INCLUDING DEFAULTS)", $fname );
 909         }
 910
 911         function listTables( $prefix = null, $fname = 'DatabasePostgres::listTables' ) {
 912                 $eschema = $this->addQuotes( $this->getCoreSchema() );
 913                 $result = $this->query( "SELECT tablename FROM pg_tables WHERE schemaname = $eschema", $fname );
 914                 $endArray = array();
 915
 916                 foreach( $result as $table ) {
 917                         $vars = get_object_vars($table);
 918                         $table = array_pop( $vars );
 919                         if( !$prefix || strpos( $table, $prefix ) === 0 ) {
 920                                 $endArray[] = $table;
 921                         }
 922                 }
 923
 924                 return $endArray;
 925         }
 926
 927         function timestamp( $ts = 0 ) {
 928                 return wfTimestamp( TS_POSTGRES, $ts );
 929         }
 930
 931         /*
 932          * Posted by cc[plus]php[at]c2se[dot]com on 25-Mar-2009 09:12
 933          * to http://www.php.net/manual/en/ref.pgsql.php
 934          *
 935          * Parsing a postgres array can be a tricky problem, he's my
 936          * take on this, it handles multi-dimensional arrays plus
 937          * escaping using a nasty regexp to determine the limits of each
 938          * data-item.
 939          *
 940          * This should really be handled by PHP PostgreSQL module
 941          *
 942          * @since 1.19
 943          * @param $text   string: postgreql array returned in a text form like {a,b}
 944          * @param $output string
 945          * @param $limit  int
 946          * @param $offset int
 947          * @return string
 948          */
 949         function pg_array_parse( $text, &$output, $limit = false, $offset = 1 ) {
 950                 if( false === $limit ) {
 951                         $limit = strlen( $text )-1;
 952                         $output = array();
 953                 }
 954                 if( '{}' == $text ) {
 955                         return $output;
 956                 }
 957                 do {
 958                         if ( '{' != $text{$offset} ) {
 959                                 preg_match( "/(\\{?\"([^\"\\\\]|\\\\.)*\"|[^,{}]+)+([,}]+)/",
 960                                         $text, $match, 0, $offset );
 961                                 $offset += strlen( $match[0] );
 962                                 $output[] = ( '"' != $match[1]{0}
 963                                                 ? $match[1]
 964                                                 : stripcslashes( substr( $match[1], 1, -1 ) ) );
 965                                 if ( '},' == $match[3] ) {
 966                                         return $output;
 967                                 }
 968                         } else {
 969                                 $offset = $this->pg_array_parse( $text, $output, $limit, $offset+1 );
 970                         }
 971                 } while ( $limit > $offset );
 972                 return $output;
 973         }
 974
 975         /**
 976          * Return aggregated value function call
 977          */
 978         function aggregateValue( $valuedata, $valuename = 'value' ) {
 979                 return $valuedata;
 980         }
 981
 982         /**
 983          * @return string wikitext of a link to the server software's web site
 984          */
 985         public static function getSoftwareLink() {
 986                 return '[http://www.postgresql.org/ PostgreSQL]';
 987         }
 988
 989
 990         /**
 991          * Return current schema (executes SELECT current_schema())
 992          * Needs transaction
 993          *
 994          * @since 1.19
 995          * @return string return default schema for the current session
 996          */
 997         function getCurrentSchema() {
 998                 $res = $this->query( "SELECT current_schema()", __METHOD__);
 999                 $row = $this->fetchRow( $res );
1000                 return $row[0];
1001         }
1002
1003         /**
1004          * Return list of schemas which are accessible without schema name
1005          * This is list does not contain magic keywords like "$user"
1006          * Needs transaction
1007          *
1008          * @seealso getSearchPath()
1009          * @seealso setSearchPath()
1010          * @since 1.19
1011          * @return array list of actual schemas for the current sesson
1012          */
1013         function getSchemas() {
1014                 $res = $this->query( "SELECT current_schemas(false)", __METHOD__);
1015                 $row = $this->fetchRow( $res );
1016                 $schemas = array();
1017                 /* PHP pgsql support does not support array type, "{a,b}" string is returned */
1018                 return $this->pg_array_parse($row[0], $schemas);
1019         }
1020
1021         /**
1022          * Return search patch for schemas
1023          * This is different from getSchemas() since it contain magic keywords
1024          * (like "$user").
1025          * Needs transaction
1026          *
1027          * @since 1.19
1028          * @return array how to search for table names schemas for the current user
1029          */
1030         function getSearchPath() {
1031                 $res = $this->query( "SHOW search_path", __METHOD__);
1032                 $row = $this->fetchRow( $res );
1033                 /* PostgreSQL returns SHOW values as strings */
1034                 return explode(",", $row[0]);
1035         }
1036
1037         /**
1038          * Update search_path, values should already be sanitized
1039          * Values may contain magic keywords like "$user"
1040          * @since 1.19
1041          *
1042          * @param $search_path array list of schemas to be searched by default
1043          */
1044         function setSearchPath( $search_path ) {
1045                 $this->query( "SET search_path = " . implode(", ", $search_path) );
1046         }
1047
1048         /**
1049          * Determine default schema for MediaWiki core
1050          * Adjust this session schema search path if desired schema exists
1051          * and is not alread there.
1052          *
1053          * We need to have name of the core schema stored to be able
1054          * to query database metadata.
1055          *
1056          * This will be also called by the installer after the schema is created
1057          *
1058          * @since 1.19
1059          * @param desired_schema string
1060          */
1061         function determineCoreSchema( $desired_schema ) {
1062                 $this->begin( __METHOD__ );
1063                 if ( $this->schemaExists( $desired_schema ) ) {
1064                         if ( in_array( $desired_schema, $this->getSchemas() ) ) {
1065                                 $this->mCoreSchema = $desired_schema;
1066                                 wfDebug("Schema \"" . $desired_schema . "\" already in the search path\n");
1067                         } else {
1068                                 /**
1069                                  * Prepend our schema (e.g. 'mediawiki') in front
1070                                  * of the search path
1071                                  * Fixes bug 15816
1072                                  */
1073                                 $search_path = $this->getSearchPath();
1074                                 array_unshift( $search_path,
1075                                         $this->addIdentifierQuotes( $desired_schema ));
1076                                 $this->setSearchPath( $search_path );
1077                                 $this->mCoreSchema = $desired_schema;
1078                                 wfDebug("Schema \"" . $desired_schema . "\" added to the search path\n");
1079                         }
1080                 } else {
1081                         $this->mCoreSchema = $this->getCurrentSchema();
1082                         wfDebug("Schema \"" . $desired_schema . "\" not found, using current \"". $this->mCoreSchema ."\"\n");
1083                 }
1084                 /* Commit SET otherwise it will be rollbacked on error or IGNORE SELECT */
1085                 $this->commit( __METHOD__ );
1086         }
1087
1088         /**
1089          * Return schema name fore core MediaWiki tables
1090          *
1091          * @since 1.19
1092          * @return string core schema name
1093          */
1094         function getCoreSchema() {
1095                 return $this->mCoreSchema;
1096         }
1097
1098         /**
1099          * @return string Version information from the database
1100          */
1101         function getServerVersion() {
1102                 if ( !isset( $this->numeric_version ) ) {
1103                         $versionInfo = pg_version( $this->mConn );
1104                         if ( version_compare( $versionInfo['client'], '7.4.0', 'lt' ) ) {
1105                                 // Old client, abort install
1106                                 $this->numeric_version = '7.3 or earlier';
1107                         } elseif ( isset( $versionInfo['server'] ) ) {
1108                                 // Normal client
1109                                 $this->numeric_version = $versionInfo['server'];
1110                         } else {
1111                                 // Bug 16937: broken pgsql extension from PHP<5.3
1112                                 $this->numeric_version = pg_parameter_status( $this->mConn, 'server_version' );
1113                         }
1114                 }
1115                 return $this->numeric_version;
1116         }
1117
1118         /**
1119          * Query whether a given relation exists (in the given schema, or the
1120          * default mw one if not given)
1121          * @return bool
1122          */
1123         function relationExists( $table, $types, $schema = false ) {
1124                 if ( !is_array( $types ) ) {
1125                         $types = array( $types );
1126                 }
1127                 if ( !$schema ) {
1128                         $schema = $this->getCoreSchema();
1129                 }
1130                 $table = $this->realTableName( $table, 'raw' );
1131                 $etable = $this->addQuotes( $table );
1132                 $eschema = $this->addQuotes( $schema );
1133                 $SQL = "SELECT 1 FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n "
1134                         . "WHERE c.relnamespace = n.oid AND c.relname = $etable AND n.nspname = $eschema "
1135                         . "AND c.relkind IN ('" . implode( "','", $types ) . "')";
1136                 $res = $this->query( $SQL );
1137                 $count = $res ? $res->numRows() : 0;
1138                 return (bool)$count;
1139         }
1140
1141         /**
1142          * For backward compatibility, this function checks both tables and
1143          * views.
1144          * @return bool
1145          */
1146         function tableExists( $table, $fname = __METHOD__, $schema = false ) {
1147                 return $this->relationExists( $table, array( 'r', 'v' ), $schema );
1148         }
1149
1150         function sequenceExists( $sequence, $schema = false ) {
1151                 return $this->relationExists( $sequence, 'S', $schema );
1152         }
1153
1154         function triggerExists( $table, $trigger ) {
1155                 $q = <<<SQL
1156         SELECT 1 FROM pg_class, pg_namespace, pg_trigger
1157                 WHERE relnamespace=pg_namespace.oid AND relkind='r'
1158                           AND tgrelid=pg_class.oid
1159                           AND nspname=%s AND relname=%s AND tgname=%s
1160 SQL;
1161                 $res = $this->query(
1162                         sprintf(
1163                                 $q,
1164                                 $this->addQuotes( $this->getCoreSchema() ),
1165                                 $this->addQuotes( $table ),
1166                                 $this->addQuotes( $trigger )
1167                         )
1168                 );
1169                 if ( !$res ) {
1170                         return null;
1171                 }
1172                 $rows = $res->numRows();
1173                 return $rows;
1174         }
1175
1176         function ruleExists( $table, $rule ) {
1177                 $exists = $this->selectField( 'pg_rules', 'rulename',
1178                         array(
1179                                 'rulename' => $rule,
1180                                 'tablename' => $table,
1181                                 'schemaname' => $this->getCoreSchema()
1182                         )
1183                 );
1184                 return $exists === $rule;
1185         }
1186
1187         function constraintExists( $table, $constraint ) {
1188                 $SQL = sprintf( "SELECT 1 FROM information_schema.table_constraints ".
1189                            "WHERE constraint_schema = %s AND table_name = %s AND constraint_name = %s",
1190                         $this->addQuotes( $this->getCoreSchema() ),
1191                         $this->addQuotes( $table ),
1192                         $this->addQuotes( $constraint )
1193                 );
1194                 $res = $this->query( $SQL );
1195                 if ( !$res ) {
1196                         return null;
1197                 }
1198                 $rows = $res->numRows();
1199                 return $rows;
1200         }
1201
1202         /**
1203          * Query whether a given schema exists. Returns true if it does, false if it doesn't.
1204          * @return bool
1205          */
1206         function schemaExists( $schema ) {
1207                 $exists = $this->selectField( '"pg_catalog"."pg_namespace"', 1,
1208                         array( 'nspname' => $schema ), __METHOD__ );
1209                 return (bool)$exists;
1210         }
1211
1212         /**
1213          * Returns true if a given role (i.e. user) exists, false otherwise.
1214          * @return bool
1215          */
1216         function roleExists( $roleName ) {
1217                 $exists = $this->selectField( '"pg_catalog"."pg_roles"', 1,
1218                         array( 'rolname' => $roleName ), __METHOD__ );
1219                 return (bool)$exists;
1220         }
1221
1222         function fieldInfo( $table, $field ) {
1223                 return PostgresField::fromText( $this, $table, $field );
1224         }
1225
1226         /**
1227          * pg_field_type() wrapper
1228          * @return string
1229          */
1230         function fieldType( $res, $index ) {
1231                 if ( $res instanceof ResultWrapper ) {
1232                         $res = $res->result;
1233                 }
1234                 return pg_field_type( $res, $index );
1235         }
1236
1237         /* Not even sure why this is used in the main codebase... */
1238         function limitResultForUpdate( $sql, $num ) {
1239                 return $sql;
1240         }
1241
1242         /**
1243          * @param $b
1244          * @return Blob
1245          */
1246         function encodeBlob( $b ) {
1247                 return new Blob( pg_escape_bytea( $this->mConn, $b ) );
1248         }
1249
1250         function decodeBlob( $b ) {
1251                 if ( $b instanceof Blob ) {
1252                         $b = $b->fetch();
1253                 }
1254                 return pg_unescape_bytea( $b );
1255         }
1256
1257         function strencode( $s ) { # Should not be called by us
1258                 return pg_escape_string( $this->mConn, $s );
1259         }
1260
1261         /**
1262          * @param $s null|bool|Blob
1263          * @return int|string
1264          */
1265         function addQuotes( $s ) {
1266                 if ( is_null( $s ) ) {
1267                         return 'NULL';
1268                 } elseif ( is_bool( $s ) ) {
1269                         return intval( $s );
1270                 } elseif ( $s instanceof Blob ) {
1271                         return "'" . $s->fetch( $s ) . "'";
1272                 }
1273                 return "'" . pg_escape_string( $this->mConn, $s ) . "'";
1274         }
1275
1276         /**
1277          * Postgres specific version of replaceVars.
1278          * Calls the parent version in Database.php
1279          *
1280          * @private
1281          *
1282          * @param $ins String: SQL string, read from a stream (usually tables.sql)
1283          *
1284          * @return string SQL string
1285          */
1286         protected function replaceVars( $ins ) {
1287                 $ins = parent::replaceVars( $ins );
1288
1289                 if ( $this->numeric_version >= 8.3 ) {
1290                         // Thanks for not providing backwards-compatibility, 8.3
1291                         $ins = preg_replace( "/to_tsvector\s*\(\s*'default'\s*,/", 'to_tsvector(', $ins );
1292                 }
1293
1294                 if ( $this->numeric_version <= 8.1 ) { // Our minimum version
1295                         $ins = str_replace( 'USING gin', 'USING gist', $ins );
1296                 }
1297
1298                 return $ins;
1299         }
1300
1301         /**
1302          * Various select options
1303          *
1304          * @private
1305          *
1306          * @param $options Array: an associative array of options to be turned into
1307          *              an SQL query, valid keys are listed in the function.
1308          * @return array
1309          */
1310         function makeSelectOptions( $options ) {
1311                 $preLimitTail = $postLimitTail = '';
1312                 $startOpts = $useIndex = '';
1313
1314                 $noKeyOptions = array();
1315                 foreach ( $options as $key => $option ) {
1316                         if ( is_numeric( $key ) ) {
1317                                 $noKeyOptions[$option] = true;
1318                         }
1319                 }
1320
1321                 if ( isset( $options['GROUP BY'] ) ) {
1322                         $gb = is_array( $options['GROUP BY'] )
1323                                 ? implode( ',', $options['GROUP BY'] )
1324                                 : $options['GROUP BY'];
1325                         $preLimitTail .= " GROUP BY {$gb}";
1326                 }
1327
1328                 if ( isset( $options['HAVING'] ) ) {
1329                         $preLimitTail .= " HAVING {$options['HAVING']}";
1330                 }
1331
1332                 if ( isset( $options['ORDER BY'] ) ) {
1333                         $ob = is_array( $options['ORDER BY'] )
1334                                 ? implode( ',', $options['ORDER BY'] )
1335                                 : $options['ORDER BY'];
1336                         $preLimitTail .= " ORDER BY {$ob}";
1337                 }
1338
1339                 //if ( isset( $options['LIMIT'] ) ) {
1340                 //      $tailOpts .= $this->limitResult( '', $options['LIMIT'],
1341                 //              isset( $options['OFFSET'] ) ? $options['OFFSET']
1342                 //              : false );
1343                 //}
1344
1345                 if ( isset( $noKeyOptions['FOR UPDATE'] ) ) {
1346                         $postLimitTail .= ' FOR UPDATE';
1347                 }
1348                 if ( isset( $noKeyOptions['LOCK IN SHARE MODE'] ) ) {
1349                         $postLimitTail .= ' LOCK IN SHARE MODE';
1350                 }
1351                 if ( isset( $noKeyOptions['DISTINCT'] ) || isset( $noKeyOptions['DISTINCTROW'] ) ) {
1352                         $startOpts .= 'DISTINCT';
1353                 }
1354
1355                 return array( $startOpts, $useIndex, $preLimitTail, $postLimitTail );
1356         }
1357
1358         function setFakeMaster( $enabled = true ) {}
1359
1360         function getDBname() {
1361                 return $this->mDBname;
1362         }
1363
1364         function getServer() {
1365                 return $this->mServer;
1366         }
1367
1368         function buildConcat( $stringList ) {
1369                 return implode( ' || ', $stringList );
1370         }
1371
1372         public function getSearchEngine() {
1373                 return 'SearchPostgres';
1374         }
1375
1376         public function streamStatementEnd( &$sql, &$newLine ) {
1377                 # Allow dollar quoting for function declarations
1378                 if ( substr( $newLine, 0, 4 ) == '$mw$' ) {
1379                         if ( $this->delimiter ) {
1380                                 $this->delimiter = false;
1381                         }
1382                         else {
1383                                 $this->delimiter = ';';
1384                         }
1385                 }
1386                 return parent::streamStatementEnd( $sql, $newLine );
1387         }
1388 } // end DatabasePostgres class