includes/DatabasePostgres.php

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