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