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