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          * Return the current value of a sequence. Assumes it has ben nextval'ed in this session.
 802          */
 803         function currentSequenceValue( $seqName ) {
 804                 $safeseq = preg_replace( "/'/", "''", $seqName );
 805                 $res = $this->query( "SELECT currval('$safeseq')" );
 806                 $row = $this->fetchRow( $res );
 807                 $currval = $row[0];
 808                 $this->freeResult( $res );
 809                 return $currval;
 810         }
 811
 812         /**
 813          * Postgres does not have a "USE INDEX" clause, so return an empty string
 814          */
 815         function useIndexClause( $index ) {
 816                 return '';
 817         }
 818
 819         # REPLACE query wrapper
 820         # Postgres simulates this with a DELETE followed by INSERT
 821         # $row is the row to insert, an associative array
 822         # $uniqueIndexes is an array of indexes. Each element may be either a
 823         # field name or an array of field names
 824         #
 825         # It may be more efficient to leave off unique indexes which are unlikely to collide.
 826         # However if you do this, you run the risk of encountering errors which wouldn't have
 827         # occurred in MySQL
 828         function replace( $table, $uniqueIndexes, $rows, $fname = 'Database::replace' ) {
 829                 $table = $this->tableName( $table );
 830
 831                 if (count($rows)==0) {
 832                         return;
 833                 }
 834
 835                 # Single row case
 836                 if ( !is_array( reset( $rows ) ) ) {
 837                         $rows = array( $rows );
 838                 }
 839
 840                 foreach( $rows as $row ) {
 841                         # Delete rows which collide
 842                         if ( $uniqueIndexes ) {
 843                                 $sql = "DELETE FROM $table WHERE ";
 844                                 $first = true;
 845                                 foreach ( $uniqueIndexes as $index ) {
 846                                         if ( $first ) {
 847                                                 $first = false;
 848                                                 $sql .= "(";
 849                                         } else {
 850                                                 $sql .= ') OR (';
 851                                         }
 852                                         if ( is_array( $index ) ) {
 853                                                 $first2 = true;
 854                                                 foreach ( $index as $col ) {
 855                                                         if ( $first2 ) {
 856                                                                 $first2 = false;
 857                                                         } else {
 858                                                                 $sql .= ' AND ';
 859                                                         }
 860                                                         $sql .= $col.'=' . $this->addQuotes( $row[$col] );
 861                                                 }
 862                                         } else {
 863                                                 $sql .= $index.'=' . $this->addQuotes( $row[$index] );
 864                                         }
 865                                 }
 866                                 $sql .= ')';
 867                                 $this->query( $sql, $fname );
 868                         }
 869
 870                         # Now insert the row
 871                         $sql = "INSERT INTO $table (" . $this->makeList( array_keys( $row ), LIST_NAMES ) .') VALUES (' .
 872                                 $this->makeList( $row, LIST_COMMA ) . ')';
 873                         $this->query( $sql, $fname );
 874                 }
 875         }
 876
 877         # DELETE where the condition is a join
 878         function deleteJoin( $delTable, $joinTable, $delVar, $joinVar, $conds, $fname = "Database::deleteJoin" ) {
 879                 if ( !$conds ) {
 880                         throw new DBUnexpectedError($this,  'Database::deleteJoin() called with empty $conds' );
 881                 }
 882
 883                 $delTable = $this->tableName( $delTable );
 884                 $joinTable = $this->tableName( $joinTable );
 885                 $sql = "DELETE FROM $delTable WHERE $delVar IN (SELECT $joinVar FROM $joinTable ";
 886                 if ( $conds != '*' ) {
 887                         $sql .= 'WHERE ' . $this->makeList( $conds, LIST_AND );
 888                 }
 889                 $sql .= ')';
 890
 891                 $this->query( $sql, $fname );
 892         }
 893
 894         # Returns the size of a text field, or -1 for "unlimited"
 895         function textFieldSize( $table, $field ) {
 896                 $table = $this->tableName( $table );
 897                 $sql = "SELECT t.typname as ftype,a.atttypmod as size
 898                         FROM pg_class c, pg_attribute a, pg_type t
 899                         WHERE relname='$table' AND a.attrelid=c.oid AND
 900                                 a.atttypid=t.oid and a.attname='$field'";
 901                 $res =$this->query($sql);
 902                 $row=$this->fetchObject($res);
 903                 if ($row->ftype=="varchar") {
 904                         $size=$row->size-4;
 905                 } else {
 906                         $size=$row->size;
 907                 }
 908                 $this->freeResult( $res );
 909                 return $size;
 910         }
 911
 912         function lowPriorityOption() {
 913                 return '';
 914         }
 915
 916         function limitResult($sql, $limit,$offset=false) {
 917                 return "$sql LIMIT $limit ".(is_numeric($offset)?" OFFSET {$offset} ":"");
 918         }
 919
 920         /**
 921          * Returns an SQL expression for a simple conditional.
 922          * Uses CASE on Postgres
 923          *
 924          * @param string $cond SQL expression which will result in a boolean value
 925          * @param string $trueVal SQL expression to return if true
 926          * @param string $falseVal SQL expression to return if false
 927          * @return string SQL fragment
 928          */
 929         function conditional( $cond, $trueVal, $falseVal ) {
 930                 return " (CASE WHEN $cond THEN $trueVal ELSE $falseVal END) ";
 931         }
 932
 933         function wasDeadlock() {
 934                 return $this->lastErrno() == '40P01';
 935         }
 936
 937         function timestamp( $ts=0 ) {
 938                 return wfTimestamp(TS_POSTGRES,$ts);
 939         }
 940
 941         /**
 942          * Return aggregated value function call
 943          */
 944         function aggregateValue ($valuedata,$valuename='value') {
 945                 return $valuedata;
 946         }
 947
 948
 949         function reportQueryError( $error, $errno, $sql, $fname, $tempIgnore = false ) {
 950                 # Ignore errors during error handling to avoid infinite recursion
 951                 $ignore = $this->ignoreErrors( true );
 952                 ++$this->mErrorCount;
 953
 954                 if ($ignore || $tempIgnore) {
 955                         wfDebug("SQL ERROR (ignored): $error\n");
 956                         $this->ignoreErrors( $ignore );
 957                 }
 958                 else {
 959                         $message = "A database error has occurred\n" .
 960                                 "Query: $sql\n" .
 961                                 "Function: $fname\n" .
 962                                 "Error: $errno $error\n";
 963                         throw new DBUnexpectedError($this, $message);
 964                 }
 965         }
 966
 967         /**
 968          * @return string wikitext of a link to the server software's web site
 969          */
 970         function getSoftwareLink() {
 971                 return "[http://www.postgresql.org/ PostgreSQL]";
 972         }
 973
 974         /**
 975          * @return string Version information from the database
 976          */
 977         function getServerVersion() {
 978                 $version = pg_fetch_result($this->doQuery("SELECT version()"),0,0);
 979                 $thisver = array();
 980                 if (!preg_match('/PostgreSQL (\d+\.\d+)(\S+)/', $version, $thisver)) {
 981                         die("Could not determine the numeric version from $version!");
 982                 }
 983                 $this->numeric_version = $thisver[1];
 984                 return $version;
 985         }
 986
 987
 988         /**
 989          * Query whether a given relation exists (in the given schema, or the
 990          * default mw one if not given)
 991          */
 992         function relationExists( $table, $types, $schema = false ) {
 993                 global $wgDBmwschema;
 994                 if (!is_array($types))
 995                         $types = array($types);
 996                 if (! $schema )
 997                         $schema = $wgDBmwschema;
 998                 $etable = $this->addQuotes($table);
 999                 $eschema = $this->addQuotes($schema);
1000                 $SQL = "SELECT 1 FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n "
1001                         . "WHERE c.relnamespace = n.oid AND c.relname = $etable AND n.nspname = $eschema "
1002                         . "AND c.relkind IN ('" . implode("','", $types) . "')";
1003                 $res = $this->query( $SQL );
1004                 $count = $res ? $res->numRows() : 0;
1005                 if ($res)
1006                         $this->freeResult( $res );
1007                 return $count ? true : false;
1008         }
1009
1010         /*
1011          * For backward compatibility, this function checks both tables and
1012          * views.
1013          */
1014         function tableExists ($table, $schema = false) {
1015                 return $this->relationExists($table, array('r', 'v'), $schema);
1016         }
1017
1018         function sequenceExists ($sequence, $schema = false) {
1019                 return $this->relationExists($sequence, 'S', $schema);
1020         }
1021
1022         function triggerExists($table, $trigger) {
1023                 global $wgDBmwschema;
1024
1025                 $q = <<<END
1026         SELECT 1 FROM pg_class, pg_namespace, pg_trigger
1027                 WHERE relnamespace=pg_namespace.oid AND relkind='r'
1028                       AND tgrelid=pg_class.oid
1029                       AND nspname=%s AND relname=%s AND tgname=%s
1030 END;
1031                 $res = $this->query(sprintf($q,
1032                                 $this->addQuotes($wgDBmwschema),
1033                                 $this->addQuotes($table),
1034                                 $this->addQuotes($trigger)));
1035                 if (!$res)
1036                         return NULL;
1037                 $rows = $res->numRows();
1038                 $this->freeResult($res);
1039                 return $rows;
1040         }
1041
1042         function ruleExists($table, $rule) {
1043                 global $wgDBmwschema;
1044                 $exists = $this->selectField("pg_rules", "rulename",
1045                                 array(  "rulename" => $rule,
1046                                         "tablename" => $table,
1047                                         "schemaname" => $wgDBmwschema));
1048                 return $exists === $rule;
1049         }
1050
1051         function constraintExists($table, $constraint) {
1052                 global $wgDBmwschema;
1053                 $SQL = sprintf("SELECT 1 FROM information_schema.table_constraints ".
1054                            "WHERE constraint_schema = %s AND table_name = %s AND constraint_name = %s",
1055                         $this->addQuotes($wgDBmwschema),
1056                         $this->addQuotes($table),
1057                         $this->addQuotes($constraint));
1058                 $res = $this->query($SQL);
1059                 if (!$res)
1060                         return NULL;
1061                 $rows = $res->numRows();
1062                 $this->freeResult($res);
1063                 return $rows;
1064         }
1065
1066         /**
1067          * Query whether a given schema exists. Returns the name of the owner
1068          */
1069         function schemaExists( $schema ) {
1070                 $eschema = preg_replace("/'/", "''", $schema);
1071                 $SQL = "SELECT rolname FROM pg_catalog.pg_namespace n, pg_catalog.pg_roles r "
1072                                 ."WHERE n.nspowner=r.oid AND n.nspname = '$eschema'";
1073                 $res = $this->query( $SQL );
1074                 if ( $res && $res->numRows() ) {
1075                         $row = $res->fetchObject();
1076                         $owner = $row->rolname;
1077                 } else {
1078                         $owner = false;
1079                 }
1080                 if ($res)
1081                         $this->freeResult($res);
1082                 return $owner;
1083         }
1084
1085         /**
1086          * Query whether a given column exists in the mediawiki schema
1087          */
1088         function fieldExists( $table, $field, $fname = 'DatabasePostgres::fieldExists' ) {
1089                 global $wgDBmwschema;
1090                 $etable = preg_replace("/'/", "''", $table);
1091                 $eschema = preg_replace("/'/", "''", $wgDBmwschema);
1092                 $ecol = preg_replace("/'/", "''", $field);
1093                 $SQL = "SELECT 1 FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n, pg_catalog.pg_attribute a "
1094                         . "WHERE c.relnamespace = n.oid AND c.relname = '$etable' AND n.nspname = '$eschema' "
1095                         . "AND a.attrelid = c.oid AND a.attname = '$ecol'";
1096                 $res = $this->query( $SQL, $fname );
1097                 $count = $res ? $res->numRows() : 0;
1098                 if ($res)
1099                         $this->freeResult( $res );
1100                 return $count;
1101         }
1102
1103         function fieldInfo( $table, $field ) {
1104                 return PostgresField::fromText($this, $table, $field);
1105         }
1106
1107         function begin( $fname = 'DatabasePostgres::begin' ) {
1108                 $this->query( 'BEGIN', $fname );
1109                 $this->mTrxLevel = 1;
1110         }
1111         function immediateCommit( $fname = 'DatabasePostgres::immediateCommit' ) {
1112                 return true;
1113         }
1114         function commit( $fname = 'DatabasePostgres::commit' ) {
1115                 $this->query( 'COMMIT', $fname );
1116                 $this->mTrxLevel = 0;
1117         }
1118
1119         /* Not even sure why this is used in the main codebase... */
1120         function limitResultForUpdate($sql, $num) {
1121                 return $sql;
1122         }
1123
1124         function setup_database() {
1125                 global $wgVersion, $wgDBmwschema, $wgDBts2schema, $wgDBport, $wgDBuser;
1126
1127                 ## Make sure that we can write to the correct schema
1128                 ## If not, Postgres will happily and silently go to the next search_path item
1129                 $ctest = "mw_test_table";
1130                 $safeschema = $this->quote_ident($wgDBmwschema);
1131                 if ($this->tableExists($ctest, $wgDBmwschema)) {
1132                         $this->doQuery("DROP TABLE $safeschema.$ctest");
1133                 }
1134                 $SQL = "CREATE TABLE $safeschema.$ctest(a int)";
1135                 $olde = error_reporting( 0 );
1136                 $res = $this->doQuery($SQL);
1137                 error_reporting( $olde );
1138                 if (!$res) {
1139                         print "<b>FAILED</b>. Make sure that the user \"$wgDBuser\" can write to the schema \"$wgDBmwschema\"</li>\n";
1140                         dieout("</ul>");
1141                 }
1142                 $this->doQuery("DROP TABLE $safeschema.mw_test_table");
1143
1144                 dbsource( "../maintenance/postgres/tables.sql", $this);
1145
1146                 ## Version-specific stuff
1147                 if ($this->numeric_version == 8.1) {
1148                         $this->doQuery("CREATE INDEX ts2_page_text ON pagecontent USING gist(textvector)");
1149                         $this->doQuery("CREATE INDEX ts2_page_title ON page USING gist(titlevector)");
1150                 }
1151                 else {
1152                         $this->doQuery("CREATE INDEX ts2_page_text ON pagecontent USING gin(textvector)");
1153                         $this->doQuery("CREATE INDEX ts2_page_title ON page USING gin(titlevector)");
1154                 }
1155
1156                 ## Update version information
1157                 $mwv = $this->addQuotes($wgVersion);
1158                 $pgv = $this->addQuotes($this->getServerVersion());
1159                 $pgu = $this->addQuotes($this->mUser);
1160                 $mws = $this->addQuotes($wgDBmwschema);
1161                 $tss = $this->addQuotes($wgDBts2schema);
1162                 $pgp = $this->addQuotes($wgDBport);
1163                 $dbn = $this->addQuotes($this->mDBname);
1164                 $ctype = pg_fetch_result($this->doQuery("SHOW lc_ctype"),0,0);
1165
1166                 $SQL = "UPDATE mediawiki_version SET mw_version=$mwv, pg_version=$pgv, pg_user=$pgu, ".
1167                                 "mw_schema = $mws, ts2_schema = $tss, pg_port=$pgp, pg_dbname=$dbn, ".
1168                                 "ctype = '$ctype' ".
1169                                 "WHERE type = 'Creation'";
1170                 $this->query($SQL);
1171
1172                 ## Avoid the non-standard "REPLACE INTO" syntax
1173                 $f = fopen( "../maintenance/interwiki.sql", 'r' );
1174                 if ($f == false ) {
1175                         dieout( "<li>Could not find the interwiki.sql file");
1176                 }
1177                 ## We simply assume it is already empty as we have just created it
1178                 $SQL = "INSERT INTO interwiki(iw_prefix,iw_url,iw_local) VALUES ";
1179                 while ( ! feof( $f ) ) {
1180                         $line = fgets($f,1024);
1181                         $matches = array();
1182                         if (!preg_match('/^\s*(\(.+?),(\d)\)/', $line, $matches)) {
1183                                 continue;
1184                         }
1185                         $this->query("$SQL $matches[1],$matches[2])");
1186                 }
1187                 print " (table interwiki successfully populated)...\n";
1188
1189                 $this->doQuery("COMMIT");
1190         }
1191
1192         function encodeBlob( $b ) {
1193                 return new Blob ( pg_escape_bytea( $b ) ) ;
1194         }
1195
1196         function decodeBlob( $b ) {
1197                 if ($b instanceof Blob) {
1198                         $b = $b->fetch();
1199                 }
1200                 return pg_unescape_bytea( $b );
1201         }
1202
1203         function strencode( $s ) { ## Should not be called by us
1204                 return pg_escape_string( $s );
1205         }
1206
1207         function addQuotes( $s ) {
1208                 if ( is_null( $s ) ) {
1209                         return 'NULL';
1210                 } else if ($s instanceof Blob) {
1211                         return "'".$s->fetch($s)."'";
1212                 }
1213                 return "'" . pg_escape_string($s) . "'";
1214         }
1215
1216         function quote_ident( $s ) {
1217                 return '"' . preg_replace( '/"/', '""', $s) . '"';
1218         }
1219
1220         /* For now, does nothing */
1221         function selectDB( $db ) {
1222                 return true;
1223         }
1224
1225         /**
1226          * Various select options
1227          *
1228          * @private
1229          *
1230          * @param array $options an associative array of options to be turned into
1231          *              an SQL query, valid keys are listed in the function.
1232          * @return array
1233          */
1234         function makeSelectOptions( $options ) {
1235                 $preLimitTail = $postLimitTail = '';
1236                 $startOpts = $useIndex = '';
1237
1238                 $noKeyOptions = array();
1239                 foreach ( $options as $key => $option ) {
1240                         if ( is_numeric( $key ) ) {
1241                                 $noKeyOptions[$option] = true;
1242                         }
1243                 }
1244
1245                 if ( isset( $options['GROUP BY'] ) ) $preLimitTail .= " GROUP BY " . $options['GROUP BY'];
1246                 if ( isset( $options['HAVING'] ) ) $preLimitTail .= " HAVING {$options['HAVING']}";
1247                 if ( isset( $options['ORDER BY'] ) ) $preLimitTail .= " ORDER BY " . $options['ORDER BY'];
1248
1249                 //if (isset($options['LIMIT'])) {
1250                 //      $tailOpts .= $this->limitResult('', $options['LIMIT'],
1251                 //              isset($options['OFFSET']) ? $options['OFFSET']
1252                 //              : false);
1253                 //}
1254
1255                 if ( isset( $noKeyOptions['FOR UPDATE'] ) ) $postLimitTail .= ' FOR UPDATE';
1256                 if ( isset( $noKeyOptions['LOCK IN SHARE MODE'] ) ) $postLimitTail .= ' LOCK IN SHARE MODE';
1257                 if ( isset( $noKeyOptions['DISTINCT'] ) || isset( $noKeyOptions['DISTINCTROW'] ) ) $startOpts .= 'DISTINCT';
1258
1259                 return array( $startOpts, $useIndex, $preLimitTail, $postLimitTail );
1260         }
1261
1262         public function setTimeout( $timeout ) {
1263                 // @todo fixme no-op
1264         }
1265
1266         function ping() {
1267                 wfDebug( "Function ping() not written for DatabasePostgres.php yet");
1268                 return true;
1269         }
1270
1271         /**
1272          * How lagged is this slave?
1273          *
1274          */
1275         public function getLag() {
1276                 # Not implemented for PostgreSQL
1277                 return false;
1278         }
1279
1280         function buildConcat( $stringList ) {
1281                 return implode( ' || ', $stringList );
1282         }
1283
1284 } // end DatabasePostgres class
1285
1286