includes/DatabasePostgres.php

   1 <?php
   2
   3 /**
   4  * This is 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  * @package MediaWiki
  11  */
  12
  13 /**
  14  * Depends on database
  15  */
  16 require_once( 'Database.php' );
  17
  18 class DatabasePostgres extends Database {
  19         var $mInsertId = NULL;
  20         var $mLastResult = NULL;
  21
  22         function DatabasePostgres($server = false, $user = false, $password = false, $dbName = false,
  23                 $failFunction = false, $flags = 0 )
  24         {
  25
  26                 global $wgOut, $wgDBprefix, $wgCommandLineMode;
  27                 # Can't get a reference if it hasn't been set yet
  28                 if ( !isset( $wgOut ) ) {
  29                         $wgOut = NULL;
  30                 }
  31                 $this->mOut =& $wgOut;
  32                 $this->mFailFunction = $failFunction;
  33                 $this->mCascadingDeletes = true;
  34                 $this->mCleanupTriggers = true;
  35                 $this->mStrictIPs = true;
  36                 $this->mFlags = $flags;
  37                 $this->open( $server, $user, $password, $dbName);
  38
  39         }
  40
  41         static function newFromParams( $server = false, $user = false, $password = false, $dbName = false,
  42                 $failFunction = false, $flags = 0)
  43         {
  44                 return new DatabasePostgres( $server, $user, $password, $dbName, $failFunction, $flags );
  45         }
  46
  47         /**
  48          * Usually aborts on failure
  49          * If the failFunction is set to a non-zero integer, returns success
  50          */
  51         function open( $server, $user, $password, $dbName ) {
  52                 # Test for Postgres support, to avoid suppressed fatal error
  53                 if ( !function_exists( 'pg_connect' ) ) {
  54                         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" );
  55                 }
  56
  57
  58                 global $wgDBport;
  59
  60                 $this->close();
  61                 $this->mServer = $server;
  62                 $port = $wgDBport;
  63                 $this->mUser = $user;
  64                 $this->mPassword = $password;
  65                 $this->mDBname = $dbName;
  66
  67                 $success = false;
  68                 $hstring="";
  69                 if ($server!=false && $server!="") {
  70                         $hstring="host=$server ";
  71                 }
  72                 if ($port!=false && $port!="") {
  73                         $hstring .= "port=$port ";
  74                 }
  75
  76                 if (!strlen($user)) { ## e.g. the class is being loaded
  77                         return;
  78                 }
  79
  80                 error_reporting( E_ALL );
  81                 @$this->mConn = pg_connect("$hstring dbname=$dbName user=$user password=$password");
  82
  83                 if ( $this->mConn == false ) {
  84                         wfDebug( "DB connection error\n" );
  85                         wfDebug( "Server: $server, Database: $dbName, User: $user, Password: " . substr( $password, 0, 3 ) . "...\n" );
  86                         wfDebug( $this->lastError()."\n" );
  87                         return false;
  88                 }
  89
  90                 $this->mOpened = true;
  91                 ## If this is the initial connection, setup the schema stuff and possibly create the user
  92                 if (defined('MEDIAWIKI_INSTALL')) {
  93                         global $wgDBname, $wgDBuser, $wgDBpass, $wgDBsuperuser, $wgDBmwschema,
  94                                 $wgDBts2schema, $wgDBts2locale;
  95                         print "OK</li>\n";
  96
  97                         print "<li>Checking the version of Postgres...";
  98                         $version = pg_fetch_result($this->doQuery("SELECT version()"),0,0);
  99                         if (!preg_match("/PostgreSQL (\d+\.\d+)(\S+)/", $version, $thisver)) {
 100                                 print "<b>FAILED</b> (could not determine the version)</li>\n";
 101                                 dieout("</ul>");
 102                         }
 103                         $PGMINVER = "8.1";
 104                         if ($thisver[1] < $PGMINVER) {
 105                                 print "<b>FAILED</b>. Required version is $PGMINVER. You have $thisver[1]$thisver[2]</li>\n";
 106                                 dieout("</ul>");
 107                         }
 108                         print "version $thisver[1]$thisver[2] is OK.</li>\n";
 109
 110                         $safeuser = $this->quote_ident($wgDBuser);
 111                         ## Are we connecting as a superuser for the first time?
 112                         if ($wgDBsuperuser) {
 113                                 ## Are we really a superuser? Check out our rights
 114                                 $SQL = "SELECT
 115                                                 CASE WHEN usesuper IS TRUE THEN
 116                                                         CASE WHEN usecreatedb IS TRUE THEN 3 ELSE 1 END
 117                                                         ELSE CASE WHEN usecreatedb IS TRUE THEN 2 ELSE 0 END
 118                         END AS rights
 119                                                 FROM pg_catalog.pg_user WHERE usename = " . $this->addQuotes($wgDBsuperuser);
 120                                 $rows = $this->numRows($res = $this->doQuery($SQL));
 121                                 if (!$rows) {
 122                                         print "<li>ERROR: Could not read permissions for user \"$wgDBsuperuser\"</li>\n";
 123                                         dieout('</ul>');
 124                                 }
 125                                 $perms = pg_fetch_result($res, 0, 0);
 126
 127                                 $SQL = "SELECT 1 FROM pg_catalog.pg_user WHERE usename = " . $this->addQuotes($wgDBuser);
 128                                 $rows = $this->numRows($this->doQuery($SQL));
 129                                 if ($rows) {
 130                                         print "<li>User \"$wgDBuser\" already exists, skipping account creation.</li>";
 131                                 }
 132                                 else {
 133                                         if ($perms != 1 and $perms != 3) {
 134                                                 print "<li>ERROR: the user \"$wgDBsuperuser\" cannot create other users. ";
 135                                                 print 'Please use a different Postgres user.</li>';
 136                                                 dieout('</ul>');
 137                                         }
 138                                         print "<li>Creating user <b>$wgDBuser</b>...";
 139                                         $safepass = $this->addQuotes($wgDBpass);
 140                                         $SQL = "CREATE USER $safeuser NOCREATEDB PASSWORD $safepass";
 141                                         $this->doQuery($SQL);
 142                                         print "OK</li>\n";
 143                                 }
 144                                 ## User now exists, check out the database
 145                                 if ($dbName != $wgDBname) {
 146                                         $SQL = "SELECT 1 FROM pg_catalog.pg_database WHERE datname = " . $this->addQuotes($wgDBname);
 147                                         $rows = $this->numRows($this->doQuery($SQL));
 148                                         if ($rows) {
 149                                                 print "<li>Database \"$wgDBname\" already exists, skipping database creation.</li>";
 150                                         }
 151                                         else {
 152                                                 if ($perms < 2) {
 153                                                         print "<li>ERROR: the user \"$wgDBsuperuser\" cannot create databases. ";
 154                                                         print 'Please use a different Postgres user.</li>';
 155                                                         dieout('</ul>');
 156                                                 }
 157                                                 print "<li>Creating database <b>$wgDBname</b>...";
 158                                                 $safename = $this->quote_ident($wgDBname);
 159                                                 $SQL = "CREATE DATABASE $safename OWNER $safeuser ";
 160                                                 $this->doQuery($SQL);
 161                                                 print "OK</li>\n";
 162                                                 ## Hopefully tsearch2 and plpgsql are in template1...
 163                                         }
 164
 165                                         ## Reconnect to check out tsearch2 rights for this user
 166                                         print "<li>Connecting to \"$wgDBname\" as superuser \"$wgDBsuperuser\" to check rights...";
 167                                         @$this->mConn = pg_connect("$hstring dbname=$wgDBname user=$user password=$password");
 168                                         if ( $this->mConn == false ) {
 169                                                 print "<b>FAILED TO CONNECT!</b></li>";
 170                                                 dieout("</ul>");
 171                                         }
 172                                         print "OK</li>\n";
 173                                 }
 174
 175                                 ## Tsearch2 checks
 176                                 print "<li>Checking that tsearch2 is installed in the database \"$wgDBname\"...";
 177                                 if (! $this->tableExists("pg_ts_cfg", $wgDBts2schema)) {
 178                                         print "<b>FAILED</b>. tsearch2 must be installed in the database \"$wgDBname\".";
 179                                         print "Please see <a href='http://www.devx.com/opensource/Article/21674/0/page/2'>this article</a>";
 180                                         print " for instructions or ask on #postgresql on irc.freenode.net</li>\n";
 181                                         dieout("</ul>");
 182                                 }
 183                                 print "OK</li>\n";
 184                                 print "<li>Ensuring that user \"$wgDBuser\" has select rights on the tsearch2 tables...";
 185                                 foreach (array('cfg','cfgmap','dict','parser') as $table) {
 186                                         $SQL = "GRANT SELECT ON pg_ts_$table TO $safeuser";
 187                                         $this->doQuery($SQL);
 188                                 }
 189                                 print "OK</li>\n";
 190
 191
 192                                 ## Setup the schema for this user if needed
 193                                 $result = $this->schemaExists($wgDBmwschema);
 194                                 $safeschema = $this->quote_ident($wgDBmwschema);
 195                                 if (!$result) {
 196                                         print "<li>Creating schema <b>$wgDBmwschema</b> ...";
 197                                         $result = $this->doQuery("CREATE SCHEMA $safeschema AUTHORIZATION $safeuser");
 198                                         if (!$result) {
 199                                                 print "<b>FAILED</b>.</li>\n";
 200                                                 dieout("</ul>");
 201                                         }
 202                                         print "OK</li>\n";
 203                                 }
 204                                 else {
 205                                         print "<li>Schema already exists, explicitly granting rights...\n";
 206                                         $safeschema2 = $this->addQuotes($wgDBmwschema);
 207                                         $SQL = "SELECT 'GRANT ALL ON '||pg_catalog.quote_ident(relname)||' TO $safeuser;'\n".
 208                                                         "FROM pg_catalog.pg_class p, pg_catalog.pg_namespace n\n".
 209                                                         "WHERE relnamespace = n.oid AND n.nspname = $safeschema2\n".
 210                                                         "AND p.relkind IN ('r','S','v')\n";
 211                                         $SQL .= "UNION\n";
 212                                         $SQL .= "SELECT 'GRANT ALL ON FUNCTION '||pg_catalog.quote_ident(proname)||'('||\n".
 213                                                         "pg_catalog.oidvectortypes(p.proargtypes)||') TO $safeuser;'\n".
 214                                                         "FROM pg_catalog.pg_proc p, pg_catalog.pg_namespace n\n".
 215                                                         "WHERE p.pronamespace = n.oid AND n.nspname = $safeschema2";
 216                                         $res = $this->doQuery($SQL);
 217                                         if (!$res) {
 218                                                 print "<b>FAILED</b>. Could not set rights for the user.</li>\n";
 219                                                 dieout("</ul>");
 220                                         }
 221                                         $this->doQuery("SET search_path = $safeschema");
 222                                         $rows = $this->numRows($res);
 223                                         while ($rows) {
 224                                                 $rows--;
 225                                                 $this->doQuery(pg_fetch_result($res, $rows, 0));
 226                                         }
 227                                         print "OK</li>";
 228                                 }
 229
 230                                 $wgDBsuperuser = '';
 231                                 return true; ## Reconnect as regular user
 232                         }
 233
 234                 if (!defined('POSTGRES_SEARCHPATH')) {
 235
 236                         ## Do we have the basic tsearch2 table?
 237                         print "<li>Checking for tsearch2 in the schema \"$wgDBts2schema\"...";
 238                         if (! $this->tableExists("pg_ts_dict", $wgDBts2schema)) {
 239                                 print "<b>FAILED</b>. Make sure tsearch2 is installed. See <a href=";
 240                                 print "'http://www.devx.com/opensource/Article/21674/0/page/2'>this article</a>";
 241                                 print " for instructions.</li>\n";
 242                                 dieout("</ul>");
 243                         }
 244                         print "OK</li>\n";
 245
 246                         ## Does this user have the rights to the tsearch2 tables?
 247                         $ctype = pg_fetch_result($this->doQuery("SHOW lc_ctype"),0,0);
 248                         print "<li>Checking tsearch2 permissions...";
 249                         $SQL = "SELECT ts_name FROM $wgDBts2schema.pg_ts_cfg WHERE locale = '$ctype'";
 250                         $SQL .= " ORDER BY CASE WHEN ts_name <> 'default' THEN 1 ELSE 0 END";
 251                         error_reporting( 0 );
 252                         $res = $this->doQuery($SQL);
 253                         error_reporting( E_ALL );
 254                         if (!$res) {
 255                                 print "<b>FAILED</b>. Make sure that the user \"$wgDBuser\" has SELECT access to the tsearch2 tables</li>\n";
 256                                 dieout("</ul>");
 257                         }
 258                         print "OK</li>";
 259
 260                         ## Will the current locale work? Can we force it to?
 261                         print "<li>Verifying tsearch2 locale with $ctype...";
 262                         $rows = $this->numRows($res);
 263                         $resetlocale = 0;
 264                         if (!$rows) {
 265                                 print "<b>not found</b></li>\n";
 266                                 print "<li>Attempting to set default tsearch2 locale to \"$ctype\"...";
 267                                 $resetlocale = 1;
 268                         }
 269                         else {
 270                                 $tsname = pg_fetch_result($res, 0, 0);
 271                                 if ($tsname != 'default') {
 272                                         print "<b>not set to default ($tsname)</b>";
 273                                         print "<li>Attempting to change tsearch2 default locale to \"$ctype\"...";
 274                                         $resetlocale = 1;
 275                                 }
 276                         }
 277                         if ($resetlocale) {
 278                                 $SQL = "UPDATE $wgDBts2schema.pg_ts_cfg SET locale = '$ctype' WHERE ts_name = 'default'";
 279                                 $res = $this->doQuery($SQL);
 280                                 if (!$res) {
 281                                         print "<b>FAILED</b>. ";
 282                                         print "Please make sure that the locale in pg_ts_cfg for \"default\" is set to \"ctype\"</li>\n";
 283                                         dieout("</ul>");
 284                                 }
 285                                 print "OK</li>";
 286                         }
 287
 288                         ## Final test: try out a simple tsearch2 query
 289                         $SQL = "SELECT $wgDBts2schema.to_tsvector('default','MediaWiki tsearch2 testing')";
 290                         $res = $this->doQuery($SQL);
 291                         if (!$res) {
 292                                 print "<b>FAILED</b>. Specifically, \"$SQL\" did not work.</li>";
 293                                 dieout("</ul>");
 294                         }
 295                         print "OK</li>";
 296
 297                         ## Do we have plpgsql installed?
 298                         print "<li>Checking for Pl/Pgsql ...";
 299                         $SQL = "SELECT 1 FROM pg_catalog.pg_language WHERE lanname = 'plpgsql'";
 300                         $rows = $this->numRows($this->doQuery($SQL));
 301                         if ($rows < 1) {
 302                                 // plpgsql is not installed, but if we have a pg_pltemplate table, we should be able to create it
 303                                 print "not installed. Attempting to install Pl/Pgsql ...";
 304                                 $SQL = "SELECT 1 FROM pg_catalog.pg_class c JOIN pg_catalog.pg_namespace n ON (n.oid = c.relnamespace) ".
 305                                         "WHERE relname = 'pg_pltemplate' AND nspname='pg_catalog'";
 306                                 $rows = $this->numRows($this->doQuery($SQL));
 307                                 if ($rows >= 1) {
 308                                         $result = $this->doQuery("CREATE LANGUAGE plpgsql");
 309                                         if (!$result) {
 310                                                 print "<b>FAILED</b>. You need to install the language plpgsql in the database <tt>$wgDBname</tt></li>";
 311                                                 dieout("</ul>");
 312                                         }
 313                                 }
 314                                 else {
 315                                         print "<b>FAILED</b>. You need to install the language plpgsql in the database <tt>$wgDBname</tt></li>";
 316                                         dieout("</ul>");
 317                                 }
 318                         }
 319                         print "OK</li>\n";
 320
 321                         ## Does the schema already exist? Who owns it?
 322                         $result = $this->schemaExists($wgDBmwschema);
 323                         if (!$result) {
 324                                 print "<li>Creating schema <b>$wgDBmwschema</b> ...";
 325                                 $result = $this->doQuery("CREATE SCHEMA $wgDBmwschema");
 326                                 if (!$result) {
 327                                         print "<b>FAILED</b>.</li>\n";
 328                                         dieout("</ul>");
 329                                 }
 330                                 print "OK</li>\n";
 331                         }
 332                         else if ($result != $user) {
 333                                 print "<li>Schema \"$wgDBmwschema\" exists but is not owned by \"$user\". Not ideal.</li>\n";
 334                         }
 335                         else {
 336                                 print "<li>Schema \"$wgDBmwschema\" exists and is owned by \"$user\". Excellent.</li>\n";
 337                         }
 338
 339                         ## Fix up the search paths if needed
 340                         print "<li>Setting the search path for user \"$user\" ...";
 341                         $path = $this->quote_ident($wgDBmwschema);
 342                         if ($wgDBts2schema !== $wgDBmwschema)
 343                                 $path .= ", ". $this->quote_ident($wgDBts2schema);
 344                         if ($wgDBmwschema !== 'public' and $wgDBts2schema !== 'public')
 345                                 $path .= ", public";
 346                         $SQL = "ALTER USER $safeuser SET search_path = $path";
 347                         $result = pg_query($this->mConn, $SQL);
 348                         if (!$result) {
 349                                 print "<b>FAILED</b>.</li>\n";
 350                                 dieout("</ul>");
 351                         }
 352                         print "OK</li>\n";
 353                         ## Set for the rest of this session
 354                         $SQL = "SET search_path = $path";
 355                         $result = pg_query($this->mConn, $SQL);
 356                         if (!$result) {
 357                                 print "<li>Failed to set search_path</li>\n";
 358                                 dieout("</ul>");
 359                         }
 360                         define( "POSTGRES_SEARCHPATH", $path );
 361                 }}
 362
 363                 global $wgCommandLineMode;
 364                 ## If called from the command-line (e.g. importDump), only show errors
 365                 if ($wgCommandLineMode) {
 366                         $this->doQuery("SET client_min_messages = 'ERROR'");
 367                 }
 368
 369                 return $this->mConn;
 370         }
 371
 372         /**
 373          * Closes a database connection, if it is open
 374          * Returns success, true if already closed
 375          */
 376         function close() {
 377                 $this->mOpened = false;
 378                 if ( $this->mConn ) {
 379                         return pg_close( $this->mConn );
 380                 } else {
 381                         return true;
 382                 }
 383         }
 384
 385         function doQuery( $sql ) {
 386                 return $this->mLastResult=pg_query( $this->mConn , $sql);
 387         }
 388
 389         function queryIgnore( $sql, $fname = '' ) {
 390                 return $this->query( $sql, $fname, true );
 391         }
 392
 393         function freeResult( $res ) {
 394                 if ( !@pg_free_result( $res ) ) {
 395                         throw new DBUnexpectedError($this,  "Unable to free Postgres result\n" );
 396                 }
 397         }
 398
 399         function fetchObject( $res ) {
 400                 @$row = pg_fetch_object( $res );
 401                 # FIXME: HACK HACK HACK HACK debug
 402
 403                 # TODO:
 404                 # hashar : not sure if the following test really trigger if the object
 405                 #          fetching failled.
 406                 if( pg_last_error($this->mConn) ) {
 407                         throw new DBUnexpectedError($this,  'SQL error: ' . htmlspecialchars( pg_last_error($this->mConn) ) );
 408                 }
 409                 return $row;
 410         }
 411
 412         function fetchRow( $res ) {
 413                 @$row = pg_fetch_array( $res );
 414                 if( pg_last_error($this->mConn) ) {
 415                         throw new DBUnexpectedError($this,  'SQL error: ' . htmlspecialchars( pg_last_error($this->mConn) ) );
 416                 }
 417                 return $row;
 418         }
 419
 420         function numRows( $res ) {
 421                 @$n = pg_num_rows( $res );
 422                 if( pg_last_error($this->mConn) ) {
 423                         throw new DBUnexpectedError($this,  'SQL error: ' . htmlspecialchars( pg_last_error($this->mConn) ) );
 424                 }
 425                 return $n;
 426         }
 427         function numFields( $res ) { return pg_num_fields( $res ); }
 428         function fieldName( $res, $n ) { return pg_field_name( $res, $n ); }
 429
 430         /**
 431          * This must be called after nextSequenceVal
 432          */
 433         function insertId() {
 434                 return $this->mInsertId;
 435         }
 436
 437         function dataSeek( $res, $row ) { return pg_result_seek( $res, $row ); }
 438         function lastError() {
 439                 if ( $this->mConn ) {
 440                         return pg_last_error();
 441                 }
 442                 else {
 443                         return "No database connection";
 444                 }
 445         }
 446         function lastErrno() {
 447                 return pg_last_error() ? 1 : 0;
 448         }
 449
 450         function affectedRows() {
 451                 return pg_affected_rows( $this->mLastResult );
 452         }
 453
 454         /**
 455          * Returns information about an index
 456          * If errors are explicitly ignored, returns NULL on failure
 457          */
 458         function indexInfo( $table, $index, $fname = 'Database::indexExists' ) {
 459                 $sql = "SELECT indexname FROM pg_indexes WHERE tablename='$table'";
 460                 $res = $this->query( $sql, $fname );
 461                 if ( !$res ) {
 462                         return NULL;
 463                 }
 464
 465                 while ( $row = $this->fetchObject( $res ) ) {
 466                         if ( $row->indexname == $index ) {
 467                                 return $row;
 468                         }
 469                 }
 470                 return false;
 471         }
 472
 473         function indexUnique ($table, $index, $fname = 'Database::indexUnique' ) {
 474                 $sql = "SELECT indexname FROM pg_indexes WHERE tablename='{$table}'".
 475                         " AND indexdef LIKE 'CREATE UNIQUE%({$index})'";
 476                 $res = $this->query( $sql, $fname );
 477                 if ( !$res )
 478                         return NULL;
 479                 while ($row = $this->fetchObject( $res ))
 480                         return true;
 481                 return false;
 482
 483         }
 484
 485         function insert( $table, $a, $fname = 'Database::insert', $options = array() ) {
 486                 # Postgres doesn't support options
 487                 # We have a go at faking one of them
 488                 # TODO: DELAYED, LOW_PRIORITY
 489
 490                 if ( !is_array($options))
 491                         $options = array($options);
 492
 493                 if ( in_array( 'IGNORE', $options ) )
 494                         $oldIgnore = $this->ignoreErrors( true );
 495
 496                 # IGNORE is performed using single-row inserts, ignoring errors in each
 497                 # FIXME: need some way to distiguish between key collision and other types of error
 498                 $oldIgnore = $this->ignoreErrors( true );
 499                 if ( !is_array( reset( $a ) ) ) {
 500                         $a = array( $a );
 501                 }
 502                 foreach ( $a as $row ) {
 503                         parent::insert( $table, $row, $fname, array() );
 504                 }
 505                 $this->ignoreErrors( $oldIgnore );
 506                 $retVal = true;
 507
 508                 if ( in_array( 'IGNORE', $options ) )
 509                         $this->ignoreErrors( $oldIgnore );
 510
 511                 return $retVal;
 512         }
 513
 514         function tableName( $name ) {
 515                 # Replace reserved words with better ones
 516                 switch( $name ) {
 517                         case 'user':
 518                                 return 'mwuser';
 519                         case 'text':
 520                                 return 'pagecontent';
 521                         default:
 522                                 return $name;
 523                 }
 524         }
 525
 526         /**
 527          * Return the next in a sequence, save the value for retrieval via insertId()
 528          */
 529         function nextSequenceValue( $seqName ) {
 530                 $safeseq = preg_replace( "/'/", "''", $seqName );
 531                 $res = $this->query( "SELECT nextval('$safeseq')" );
 532                 $row = $this->fetchRow( $res );
 533                 $this->mInsertId = $row[0];
 534                 $this->freeResult( $res );
 535                 return $this->mInsertId;
 536         }
 537
 538         /**
 539          * Postgres does not have a "USE INDEX" clause, so return an empty string
 540          */
 541         function useIndexClause( $index ) {
 542                 return '';
 543         }
 544
 545         # REPLACE query wrapper
 546         # Postgres simulates this with a DELETE followed by INSERT
 547         # $row is the row to insert, an associative array
 548         # $uniqueIndexes is an array of indexes. Each element may be either a
 549         # field name or an array of field names
 550         #
 551         # It may be more efficient to leave off unique indexes which are unlikely to collide.
 552         # However if you do this, you run the risk of encountering errors which wouldn't have
 553         # occurred in MySQL
 554         function replace( $table, $uniqueIndexes, $rows, $fname = 'Database::replace' ) {
 555                 $table = $this->tableName( $table );
 556
 557                 if (count($rows)==0) {
 558                         return;
 559                 }
 560
 561                 # Single row case
 562                 if ( !is_array( reset( $rows ) ) ) {
 563                         $rows = array( $rows );
 564                 }
 565
 566                 foreach( $rows as $row ) {
 567                         # Delete rows which collide
 568                         if ( $uniqueIndexes ) {
 569                                 $sql = "DELETE FROM $table WHERE ";
 570                                 $first = true;
 571                                 foreach ( $uniqueIndexes as $index ) {
 572                                         if ( $first ) {
 573                                                 $first = false;
 574                                                 $sql .= "(";
 575                                         } else {
 576                                                 $sql .= ') OR (';
 577                                         }
 578                                         if ( is_array( $index ) ) {
 579                                                 $first2 = true;
 580                                                 foreach ( $index as $col ) {
 581                                                         if ( $first2 ) {
 582                                                                 $first2 = false;
 583                                                         } else {
 584                                                                 $sql .= ' AND ';
 585                                                         }
 586                                                         $sql .= $col.'=' . $this->addQuotes( $row[$col] );
 587                                                 }
 588                                         } else {
 589                                                 $sql .= $index.'=' . $this->addQuotes( $row[$index] );
 590                                         }
 591                                 }
 592                                 $sql .= ')';
 593                                 $this->query( $sql, $fname );
 594                         }
 595
 596                         # Now insert the row
 597                         $sql = "INSERT INTO $table (" . $this->makeList( array_keys( $row ), LIST_NAMES ) .') VALUES (' .
 598                                 $this->makeList( $row, LIST_COMMA ) . ')';
 599                         $this->query( $sql, $fname );
 600                 }
 601         }
 602
 603         # DELETE where the condition is a join
 604         function deleteJoin( $delTable, $joinTable, $delVar, $joinVar, $conds, $fname = "Database::deleteJoin" ) {
 605                 if ( !$conds ) {
 606                         throw new DBUnexpectedError($this,  'Database::deleteJoin() called with empty $conds' );
 607                 }
 608
 609                 $delTable = $this->tableName( $delTable );
 610                 $joinTable = $this->tableName( $joinTable );
 611                 $sql = "DELETE FROM $delTable WHERE $delVar IN (SELECT $joinVar FROM $joinTable ";
 612                 if ( $conds != '*' ) {
 613                         $sql .= 'WHERE ' . $this->makeList( $conds, LIST_AND );
 614                 }
 615                 $sql .= ')';
 616
 617                 $this->query( $sql, $fname );
 618         }
 619
 620         # Returns the size of a text field, or -1 for "unlimited"
 621         function textFieldSize( $table, $field ) {
 622                 $table = $this->tableName( $table );
 623                 $sql = "SELECT t.typname as ftype,a.atttypmod as size
 624                         FROM pg_class c, pg_attribute a, pg_type t
 625                         WHERE relname='$table' AND a.attrelid=c.oid AND
 626                                 a.atttypid=t.oid and a.attname='$field'";
 627                 $res =$this->query($sql);
 628                 $row=$this->fetchObject($res);
 629                 if ($row->ftype=="varchar") {
 630                         $size=$row->size-4;
 631                 } else {
 632                         $size=$row->size;
 633                 }
 634                 $this->freeResult( $res );
 635                 return $size;
 636         }
 637
 638         function lowPriorityOption() {
 639                 return '';
 640         }
 641
 642         function limitResult($sql, $limit,$offset) {
 643                 return "$sql LIMIT $limit ".(is_numeric($offset)?" OFFSET {$offset} ":"");
 644         }
 645
 646         /**
 647          * Returns an SQL expression for a simple conditional.
 648          * Uses CASE on Postgres
 649          *
 650          * @param string $cond SQL expression which will result in a boolean value
 651          * @param string $trueVal SQL expression to return if true
 652          * @param string $falseVal SQL expression to return if false
 653          * @return string SQL fragment
 654          */
 655         function conditional( $cond, $trueVal, $falseVal ) {
 656                 return " (CASE WHEN $cond THEN $trueVal ELSE $falseVal END) ";
 657         }
 658
 659         # FIXME: actually detecting deadlocks might be nice
 660         function wasDeadlock() {
 661                 return false;
 662         }
 663
 664         function timestamp( $ts=0 ) {
 665                 return wfTimestamp(TS_POSTGRES,$ts);
 666         }
 667
 668         /**
 669          * Return aggregated value function call
 670          */
 671         function aggregateValue ($valuedata,$valuename='value') {
 672                 return $valuedata;
 673         }
 674
 675
 676         function reportQueryError( $error, $errno, $sql, $fname, $tempIgnore = false ) {
 677                 $message = "A database error has occurred\n" .
 678                         "Query: $sql\n" .
 679                         "Function: $fname\n" .
 680                         "Error: $errno $error\n";
 681                 throw new DBUnexpectedError($this, $message);
 682         }
 683
 684         /**
 685          * @return string wikitext of a link to the server software's web site
 686          */
 687         function getSoftwareLink() {
 688                 return "[http://www.postgresql.org/ PostgreSQL]";
 689         }
 690
 691         /**
 692          * @return string Version information from the database
 693          */
 694         function getServerVersion() {
 695                 $res = $this->query( "SELECT version()" );
 696                 $row = $this->fetchRow( $res );
 697                 $version = $row[0];
 698                 $this->freeResult( $res );
 699                 return $version;
 700         }
 701
 702
 703         /**
 704          * Query whether a given table exists (in the given schema, or the default mw one if not given)
 705          */
 706         function tableExists( $table, $schema = false ) {
 707                 global $wgDBmwschema;
 708                 if (! $schema )
 709                         $schema = $wgDBmwschema;
 710                 $etable = preg_replace("/'/", "''", $table);
 711                 $eschema = preg_replace("/'/", "''", $schema);
 712                 $SQL = "SELECT 1 FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n "
 713                         . "WHERE c.relnamespace = n.oid AND c.relname = '$etable' AND n.nspname = '$eschema' "
 714                         . "AND c.relkind IN ('r','v')";
 715                 $res = $this->query( $SQL );
 716                 $count = $res ? pg_num_rows($res) : 0;
 717                 if ($res)
 718                         $this->freeResult( $res );
 719                 return $count;
 720         }
 721
 722
 723         /**
 724          * Query whether a given schema exists. Returns the name of the owner
 725          */
 726         function schemaExists( $schema ) {
 727                 $eschema = preg_replace("/'/", "''", $schema);
 728                 $SQL = "SELECT rolname FROM pg_catalog.pg_namespace n, pg_catalog.pg_roles r "
 729                                 ."WHERE n.nspowner=r.oid AND n.nspname = '$eschema'";
 730                 $res = $this->query( $SQL );
 731                 $owner = $res ? pg_num_rows($res) ? pg_fetch_result($res, 0, 0) : false : false;
 732                 if ($res)
 733                         $this->freeResult($res);
 734                 return $owner;
 735         }
 736
 737         /**
 738          * Query whether a given column exists in the mediawiki schema
 739          */
 740         function fieldExists( $table, $field ) {
 741                 global $wgDBmwschema;
 742                 $etable = preg_replace("/'/", "''", $table);
 743                 $eschema = preg_replace("/'/", "''", $wgDBmwschema);
 744                 $ecol = preg_replace("/'/", "''", $field);
 745                 $SQL = "SELECT 1 FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n, pg_catalog.pg_attribute a "
 746                         . "WHERE c.relnamespace = n.oid AND c.relname = '$etable' AND n.nspname = '$eschema' "
 747                         . "AND a.attrelid = c.oid AND a.attname = '$ecol'";
 748                 $res = $this->query( $SQL );
 749                 $count = $res ? pg_num_rows($res) : 0;
 750                 if ($res)
 751                         $this->freeResult( $res );
 752                 return $count;
 753         }
 754
 755         function fieldInfo( $table, $field ) {
 756                 $res = $this->query( "SELECT $field FROM $table LIMIT 1" );
 757                 $type = pg_field_type( $res, 0 );
 758                 return $type;
 759         }
 760
 761         function begin( $fname = 'DatabasePostgrs::begin' ) {
 762                 $this->query( 'BEGIN', $fname );
 763                 $this->mTrxLevel = 1;
 764         }
 765         function immediateCommit( $fname = 'DatabasePostgres::immediateCommit' ) {
 766                 return true;
 767         }
 768         function commit( $fname = 'DatabasePostgres::commit' ) {
 769                 $this->query( 'COMMIT', $fname );
 770                 $this->mTrxLevel = 0;
 771         }
 772
 773         /* Not even sure why this is used in the main codebase... */
 774         function limitResultForUpdate($sql, $num) {
 775                 return $sql;
 776         }
 777
 778         function setup_database() {
 779                 global $wgVersion, $wgDBmwschema, $wgDBts2schema, $wgDBport;
 780
 781                 dbsource( "../maintenance/postgres/tables.sql", $this);
 782
 783                 ## Update version information
 784                 $mwv = $this->addQuotes($wgVersion);
 785                 $pgv = $this->addQuotes($this->getServerVersion());
 786                 $pgu = $this->addQuotes($this->mUser);
 787                 $mws = $this->addQuotes($wgDBmwschema);
 788                 $tss = $this->addQuotes($wgDBts2schema);
 789                 $pgp = $this->addQuotes($wgDBport);
 790                 $dbn = $this->addQuotes($this->mDBname);
 791                 $ctype = pg_fetch_result($this->doQuery("SHOW lc_ctype"),0,0);
 792
 793                 $SQL = "UPDATE mediawiki_version SET mw_version=$mwv, pg_version=$pgv, pg_user=$pgu, ".
 794                                 "mw_schema = $mws, ts2_schema = $tss, pg_port=$pgp, pg_dbname=$dbn, ".
 795                                 "ctype = '$ctype' ".
 796                                 "WHERE type = 'Creation'";
 797                 $this->query($SQL);
 798
 799                 ## Avoid the non-standard "REPLACE INTO" syntax
 800                 $f = fopen( "../maintenance/interwiki.sql", 'r' );
 801                 if ($f == false ) {
 802                         dieout( "<li>Could not find the interwiki.sql file");
 803                 }
 804                 ## We simply assume it is already empty as we have just created it
 805                 $SQL = "INSERT INTO interwiki(iw_prefix,iw_url,iw_local) VALUES ";
 806                 while ( ! feof( $f ) ) {
 807                         $line = fgets($f,1024);
 808                         if (!preg_match("/^\s*(\(.+?),(\d)\)/", $line, $matches)) {
 809                                 continue;
 810                         }
 811                         $yesno = $matches[2]; ## ? "'true'" : "'false'";
 812                         $this->query("$SQL $matches[1],$matches[2])");
 813                 }
 814                 print " (table interwiki successfully populated)...\n";
 815         }
 816
 817         function encodeBlob($b) {
 818                 return array('bytea',pg_escape_bytea($b));
 819         }
 820         function decodeBlob($b) {
 821                 return pg_unescape_bytea( $b );
 822         }
 823
 824         function strencode( $s ) { ## Should not be called by us
 825                 return pg_escape_string( $s );
 826         }
 827
 828         function addQuotes( $s ) {
 829                 if ( is_null( $s ) ) {
 830                         return 'NULL';
 831                 } else if (is_array( $s )) { ## Assume it is bytea data
 832                         return "E'$s[1]'";
 833                 }
 834                 return "'" . pg_escape_string($s) . "'";
 835                 return "E'" . pg_escape_string($s) . "'";
 836         }
 837
 838         function quote_ident( $s ) {
 839                 return '"' . preg_replace( '/"/', '""', $s) . '"';
 840         }
 841
 842 }
 843
 844 ?>