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