includes/DatabasePostgres.php

   1 <?php
   2
   3 /**
   4  * This is PostgreSQL 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->mFlags = $flags;
  34
  35                 $this->open( $server, $user, $password, $dbName);
  36
  37         }
  38
  39         static function newFromParams( $server = false, $user = false, $password = false, $dbName = false,
  40                 $failFunction = false, $flags = 0)
  41         {
  42                 return new DatabasePostgres( $server, $user, $password, $dbName, $failFunction, $flags );
  43         }
  44
  45         /**
  46          * Usually aborts on failure
  47          * If the failFunction is set to a non-zero integer, returns success
  48          */
  49         function open( $server, $user, $password, $dbName ) {
  50                 # Test for PostgreSQL support, to avoid suppressed fatal error
  51                 if ( !function_exists( 'pg_connect' ) ) {
  52                         throw new DBConnectionError( $this, "PostgreSQL functions missing, have you compiled PHP with the --with-pgsql option?\n" );
  53                 }
  54
  55                 global $wgDBport;
  56
  57                 $this->close();
  58                 $this->mServer = $server;
  59                 $port = $wgDBport;
  60                 $this->mUser = $user;
  61                 $this->mPassword = $password;
  62                 $this->mDBname = $dbName;
  63
  64                 $success = false;
  65
  66                 $hstring="";
  67                 if ($server!=false && $server!="") {
  68                         $hstring="host=$server ";
  69                 }
  70                 if ($port!=false && $port!="") {
  71                         $hstring .= "port=$port ";
  72                 }
  73
  74                 error_reporting( E_ALL );
  75
  76                 @$this->mConn = pg_connect("$hstring dbname=$dbName user=$user password=$password");
  77
  78                 if ( $this->mConn == false ) {
  79                         wfDebug( "DB connection error\n" );
  80                         wfDebug( "Server: $server, Database: $dbName, User: $user, Password: " . substr( $password, 0, 3 ) . "...\n" );
  81                         wfDebug( $this->lastError()."\n" );
  82                         return false;
  83                 }
  84
  85                 $this->mOpened = true;
  86                 ## If this is the initial connection, setup the schema stuff
  87                 if (defined('MEDIAWIKI_INSTALL') and !defined('POSTGRES_SEARCHPATH')) {
  88                         global $wgDBmwschema, $wgDBts2schema, $wgDBname;
  89
  90                         ## Do we have the basic tsearch2 table?
  91                         print "<li>Checking for tsearch2 ...";
  92                         if (! $this->tableExists("pg_ts_dict", $wgDBts2schema)) {
  93                                 print "<b>FAILED</b>. Make sure tsearch2 is installed. See <a href=";
  94                                 print "'http://www.devx.com/opensource/Article/21674/0/page/2'>this article</a>";
  95                                 print " for instructions.</li>\n";
  96                                 dieout("</ul>");
  97                         }
  98                         print "OK</li>\n";
  99
 100                         ## Do we have plpgsql installed?
 101                         print "<li>Checking for plpgsql ...";
 102                         $SQL = "SELECT 1 FROM pg_catalog.pg_language WHERE lanname = 'plpgsql'";
 103                         $res = $this->doQuery($SQL);
 104                         $rows = $this->numRows($this->doQuery($SQL));
 105                         if ($rows < 1) {
 106                                 print "<b>FAILED</b>. Make sure the language plpgsql is installed for the database <tt>$wgDBname</tt>t</li>";
 107                                 ## XXX Better help
 108                                 dieout("</ul>");
 109                         }
 110                         print "OK</li>\n";
 111
 112                         ## Does the schema already exist? Who owns it?
 113                         $result = $this->schemaExists($wgDBmwschema);
 114                         if (!$result) {
 115                                 print "<li>Creating schema <b>$wgDBmwschema</b> ...";
 116                                 $result = $this->doQuery("CREATE SCHEMA $wgDBmwschema");
 117                                 if (!$result) {
 118                                         print "FAILED.</li>\n";
 119                                         return false;
 120                                 }
 121                                 print "ok</li>\n";
 122                         }
 123                         else if ($result != $user) {
 124                                 print "<li>Schema <b>$wgDBmwschema</b> exists but is not owned by <b>$user</b>. Not ideal.</li>\n";
 125                         }
 126                         else {
 127                                 print "<li>Schema <b>$wgDBmwschema</b> exists and is owned by <b>$user ($result)</b>. Excellent.</li>\n";
 128                         }
 129
 130                         ## Fix up the search paths if needed
 131                         print "<li>Setting the search path for user <b>$user</b> ...";
 132                         $path = "$wgDBmwschema";
 133                         if ($wgDBts2schema !== $wgDBmwschema)
 134                                 $path .= ", $wgDBts2schema";
 135                         if ($wgDBmwschema !== 'public' and $wgDBts2schema !== 'public')
 136                                 $path .= ", public";
 137                         $SQL = "ALTER USER $user SET search_path = $path";
 138                         $result = pg_query($this->mConn, $SQL);
 139                         if (!$result) {
 140                                 print "FAILED.</li>\n";
 141                                 return false;
 142                         }
 143                         print "ok</li>\n";
 144                         ## Set for the rest of this session
 145                         $SQL = "SET search_path = $path";
 146                         $result = pg_query($this->mConn, $SQL);
 147                         if (!$result) {
 148                                 print "<li>Failed to set search_path</li>\n";
 149                                 return false;
 150                         }
 151                         define( "POSTGRES_SEARCHPATH", $path );
 152                 }
 153
 154                 return $this->mConn;
 155         }
 156
 157         /**
 158          * Closes a database connection, if it is open
 159          * Returns success, true if already closed
 160          */
 161         function close() {
 162                 $this->mOpened = false;
 163                 if ( $this->mConn ) {
 164                         return pg_close( $this->mConn );
 165                 } else {
 166                         return true;
 167                 }
 168         }
 169
 170         function doQuery( $sql ) {
 171                 return $this->mLastResult=pg_query( $this->mConn , $sql);
 172         }
 173
 174         function queryIgnore( $sql, $fname = '' ) {
 175                 return $this->query( $sql, $fname, true );
 176         }
 177
 178         function freeResult( $res ) {
 179                 if ( !@pg_free_result( $res ) ) {
 180                         throw new DBUnexpectedError($this,  "Unable to free PostgreSQL result\n" );
 181                 }
 182         }
 183
 184         function fetchObject( $res ) {
 185                 @$row = pg_fetch_object( $res );
 186                 # FIXME: HACK HACK HACK HACK debug
 187
 188                 # TODO:
 189                 # hashar : not sure if the following test really trigger if the object
 190                 #          fetching failled.
 191                 if( pg_last_error($this->mConn) ) {
 192                         throw new DBUnexpectedError($this,  'SQL error: ' . htmlspecialchars( pg_last_error($this->mConn) ) );
 193                 }
 194                 return $row;
 195         }
 196
 197         function fetchRow( $res ) {
 198                 @$row = pg_fetch_array( $res );
 199                 if( pg_last_error($this->mConn) ) {
 200                         throw new DBUnexpectedError($this,  'SQL error: ' . htmlspecialchars( pg_last_error($this->mConn) ) );
 201                 }
 202                 return $row;
 203         }
 204
 205         function numRows( $res ) {
 206                 @$n = pg_num_rows( $res );
 207                 if( pg_last_error($this->mConn) ) {
 208                         throw new DBUnexpectedError($this,  'SQL error: ' . htmlspecialchars( pg_last_error($this->mConn) ) );
 209                 }
 210                 return $n;
 211         }
 212         function numFields( $res ) { return pg_num_fields( $res ); }
 213         function fieldName( $res, $n ) { return pg_field_name( $res, $n ); }
 214
 215         /**
 216          * This must be called after nextSequenceVal
 217          */
 218         function insertId() {
 219                 return $this->mInsertId;
 220         }
 221
 222         function dataSeek( $res, $row ) { return pg_result_seek( $res, $row ); }
 223         function lastError() {
 224                 if ( $this->mConn ) {
 225                         return pg_last_error();
 226                 }
 227                 else {
 228                         return "No database connection";
 229                 }
 230         }
 231         function lastErrno() { return 1; }
 232
 233         function affectedRows() {
 234                 return pg_affected_rows( $this->mLastResult );
 235         }
 236
 237         /**
 238          * Returns information about an index
 239          * If errors are explicitly ignored, returns NULL on failure
 240          */
 241         function indexInfo( $table, $index, $fname = 'Database::indexExists' ) {
 242                 $sql = "SELECT indexname FROM pg_indexes WHERE tablename='$table'";
 243                 $res = $this->query( $sql, $fname );
 244                 if ( !$res ) {
 245                         return NULL;
 246                 }
 247
 248                 while ( $row = $this->fetchObject( $res ) ) {
 249                         if ( $row->indexname == $index ) {
 250                                 return $row;
 251                         }
 252                 }
 253                 return false;
 254         }
 255
 256         function indexUnique ($table, $index, $fname = 'Database::indexUnique' ) {
 257                 $sql = "SELECT indexname FROM pg_indexes WHERE tablename='{$table}'".
 258                         " AND indexdef LIKE 'CREATE UNIQUE%({$index})'";
 259                 $res = $this->query( $sql, $fname );
 260                 if ( !$res )
 261                         return NULL;
 262                 while ($row = $this->fetchObject( $res ))
 263                         return true;
 264                 return false;
 265
 266         }
 267
 268         function insert( $table, $a, $fname = 'Database::insert', $options = array() ) {
 269                 # PostgreSQL doesn't support options
 270                 # We have a go at faking one of them
 271                 # TODO: DELAYED, LOW_PRIORITY
 272
 273                 if ( !is_array($options))
 274                         $options = array($options);
 275
 276                 if ( in_array( 'IGNORE', $options ) )
 277                         $oldIgnore = $this->ignoreErrors( true );
 278
 279                 # IGNORE is performed using single-row inserts, ignoring errors in each
 280                 # FIXME: need some way to distiguish between key collision and other types of error
 281                 $oldIgnore = $this->ignoreErrors( true );
 282                 if ( !is_array( reset( $a ) ) ) {
 283                         $a = array( $a );
 284                 }
 285                 foreach ( $a as $row ) {
 286                         parent::insert( $table, $row, $fname, array() );
 287                 }
 288                 $this->ignoreErrors( $oldIgnore );
 289                 $retVal = true;
 290
 291                 if ( in_array( 'IGNORE', $options ) )
 292                         $this->ignoreErrors( $oldIgnore );
 293
 294                 return $retVal;
 295         }
 296
 297         function tableName( $name ) {
 298                 # Replace backticks into double quotes
 299                 $name = strtr($name,'`','"');
 300
 301                 # Now quote PG reserved keywords
 302                 switch( $name ) {
 303                         case 'user':
 304                         case 'old':
 305                         case 'group':
 306                                 return '"' . $name . '"';
 307
 308                         default:
 309                                 return $name;
 310                 }
 311         }
 312
 313         /**
 314          * Return the next in a sequence, save the value for retrieval via insertId()
 315          */
 316         function nextSequenceValue( $seqName ) {
 317                 $safeseq = preg_replace( "/'/", "''", $seqName );
 318                 $res = $this->query( "SELECT nextval('$safeseq')" );
 319                 $row = $this->fetchRow( $res );
 320                 $this->mInsertId = $row[0];
 321                 $this->freeResult( $res );
 322                 return $this->mInsertId;
 323         }
 324
 325         /**
 326          * USE INDEX clause
 327          * PostgreSQL doesn't have them and returns ""
 328          */
 329         function useIndexClause( $index ) {
 330                 return '';
 331         }
 332
 333         # REPLACE query wrapper
 334         # PostgreSQL simulates this with a DELETE followed by INSERT
 335         # $row is the row to insert, an associative array
 336         # $uniqueIndexes is an array of indexes. Each element may be either a
 337         # field name or an array of field names
 338         #
 339         # It may be more efficient to leave off unique indexes which are unlikely to collide.
 340         # However if you do this, you run the risk of encountering errors which wouldn't have
 341         # occurred in MySQL
 342         function replace( $table, $uniqueIndexes, $rows, $fname = 'Database::replace' ) {
 343                 $table = $this->tableName( $table );
 344
 345                 if (count($rows)==0) {
 346                         return;
 347                 }
 348
 349                 # Single row case
 350                 if ( !is_array( reset( $rows ) ) ) {
 351                         $rows = array( $rows );
 352                 }
 353
 354                 foreach( $rows as $row ) {
 355                         # Delete rows which collide
 356                         if ( $uniqueIndexes ) {
 357                                 $sql = "DELETE FROM $table WHERE ";
 358                                 $first = true;
 359                                 foreach ( $uniqueIndexes as $index ) {
 360                                         if ( $first ) {
 361                                                 $first = false;
 362                                                 $sql .= "(";
 363                                         } else {
 364                                                 $sql .= ') OR (';
 365                                         }
 366                                         if ( is_array( $index ) ) {
 367                                                 $first2 = true;
 368                                                 foreach ( $index as $col ) {
 369                                                         if ( $first2 ) {
 370                                                                 $first2 = false;
 371                                                         } else {
 372                                                                 $sql .= ' AND ';
 373                                                         }
 374                                                         $sql .= $col.'=' . $this->addQuotes( $row[$col] );
 375                                                 }
 376                                         } else {
 377                                                 $sql .= $index.'=' . $this->addQuotes( $row[$index] );
 378                                         }
 379                                 }
 380                                 $sql .= ')';
 381                                 $this->query( $sql, $fname );
 382                         }
 383
 384                         # Now insert the row
 385                         $sql = "INSERT INTO $table (" . $this->makeList( array_keys( $row ), LIST_NAMES ) .') VALUES (' .
 386                                 $this->makeList( $row, LIST_COMMA ) . ')';
 387                         $this->query( $sql, $fname );
 388                 }
 389         }
 390
 391         # DELETE where the condition is a join
 392         function deleteJoin( $delTable, $joinTable, $delVar, $joinVar, $conds, $fname = "Database::deleteJoin" ) {
 393                 if ( !$conds ) {
 394                         throw new DBUnexpectedError($this,  'Database::deleteJoin() called with empty $conds' );
 395                 }
 396
 397                 $delTable = $this->tableName( $delTable );
 398                 $joinTable = $this->tableName( $joinTable );
 399                 $sql = "DELETE FROM $delTable WHERE $delVar IN (SELECT $joinVar FROM $joinTable ";
 400                 if ( $conds != '*' ) {
 401                         $sql .= 'WHERE ' . $this->makeList( $conds, LIST_AND );
 402                 }
 403                 $sql .= ')';
 404
 405                 $this->query( $sql, $fname );
 406         }
 407
 408         # Returns the size of a text field, or -1 for "unlimited"
 409         function textFieldSize( $table, $field ) {
 410                 $table = $this->tableName( $table );
 411                 $sql = "SELECT t.typname as ftype,a.atttypmod as size
 412                         FROM pg_class c, pg_attribute a, pg_type t
 413                         WHERE relname='$table' AND a.attrelid=c.oid AND
 414                                 a.atttypid=t.oid and a.attname='$field'";
 415                 $res =$this->query($sql);
 416                 $row=$this->fetchObject($res);
 417                 if ($row->ftype=="varchar") {
 418                         $size=$row->size-4;
 419                 } else {
 420                         $size=$row->size;
 421                 }
 422                 $this->freeResult( $res );
 423                 return $size;
 424         }
 425
 426         function lowPriorityOption() {
 427                 return '';
 428         }
 429
 430         function limitResult($sql, $limit,$offset) {
 431                 return "$sql LIMIT $limit ".(is_numeric($offset)?" OFFSET {$offset} ":"");
 432         }
 433
 434         /**
 435          * Returns an SQL expression for a simple conditional.
 436          * Uses CASE on PostgreSQL.
 437          *
 438          * @param string $cond SQL expression which will result in a boolean value
 439          * @param string $trueVal SQL expression to return if true
 440          * @param string $falseVal SQL expression to return if false
 441          * @return string SQL fragment
 442          */
 443         function conditional( $cond, $trueVal, $falseVal ) {
 444                 return " (CASE WHEN $cond THEN $trueVal ELSE $falseVal END) ";
 445         }
 446
 447         # FIXME: actually detecting deadlocks might be nice
 448         function wasDeadlock() {
 449                 return false;
 450         }
 451
 452         # Return DB-style timestamp used for MySQL schema
 453         function timestamp( $ts=0 ) {
 454                 return wfTimestamp(TS_DB,$ts);
 455         }
 456
 457         /**
 458          * Return aggregated value function call
 459          */
 460         function aggregateValue ($valuedata,$valuename='value') {
 461                 return $valuedata;
 462         }
 463
 464
 465         function reportQueryError( $error, $errno, $sql, $fname, $tempIgnore = false ) {
 466                 $message = "A database error has occurred\n" .
 467                         "Query: $sql\n" .
 468                         "Function: $fname\n" .
 469                         "Error: $errno $error\n";
 470                 throw new DBUnexpectedError($this, $message);
 471         }
 472
 473         /**
 474          * @return string wikitext of a link to the server software's web site
 475          */
 476         function getSoftwareLink() {
 477                 return "[http://www.postgresql.org/ PostgreSQL]";
 478         }
 479
 480         /**
 481          * @return string Version information from the database
 482          */
 483         function getServerVersion() {
 484                 $res = $this->query( "SELECT version()" );
 485                 $row = $this->fetchRow( $res );
 486                 $version = $row[0];
 487                 $this->freeResult( $res );
 488                 return $version;
 489         }
 490
 491
 492         /**
 493          * Query whether a given table exists (in the given schema, or the default mw one if not given)
 494          */
 495         function tableExists( $table, $schema = false ) {
 496                 global $wgDBmwschema;
 497                 if (! $schema )
 498                         $schema = $wgDBmwschema;
 499                 $etable = preg_replace("/'/", "''", $table);
 500                 $eschema = preg_replace("/'/", "''", $schema);
 501                 $SQL = "SELECT 1 FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n "
 502                         . "WHERE c.relnamespace = n.oid AND c.relname = '$etable' AND n.nspname = '$eschema'";
 503                 $res = $this->query( $SQL );
 504                 $count = $res ? pg_num_rows($res) : 0;
 505                 if ($res)
 506                         $this->freeResult( $res );
 507                 return $count;
 508         }
 509
 510
 511         /**
 512          * Query whether a given schema exists. Returns the name of the owner
 513          */
 514         function schemaExists( $schema ) {
 515                 $eschema = preg_replace("/'/", "''", $schema);
 516                 $SQL = "SELECT rolname FROM pg_catalog.pg_namespace n, pg_catalog.pg_roles r "
 517                                 ."WHERE n.nspowner=r.oid AND n.nspname = '$eschema'";
 518                 $res = $this->query( $SQL );
 519                 $owner = $res ? pg_num_rows($res) ? pg_fetch_result($res, 0, 0) : false : false;
 520                 if ($res)
 521                         $this->freeResult($res);
 522                 return $owner;
 523         }
 524
 525         /**
 526          * Query whether a given column exists in the mediawiki schema
 527          */
 528         function fieldExists( $table, $field ) {
 529                 global $wgDBmwschema;
 530                 $etable = preg_replace("/'/", "''", $table);
 531                 $eschema = preg_replace("/'/", "''", $wgDBmwschema);
 532                 $ecol = preg_replace("/'/", "''", $field);
 533                 $SQL = "SELECT 1 FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n, pg_catalog.pg_attribute a "
 534                         . "WHERE c.relnamespace = n.oid AND c.relname = '$etable' AND n.nspname = '$eschema' "
 535                         . "AND a.attrelid = c.oid AND a.attname = '$ecol'";
 536                 $res = $this->query( $SQL );
 537                 $count = $res ? pg_num_rows($res) : 0;
 538                 if ($res)
 539                         $this->freeResult( $res );
 540                 return $count;
 541         }
 542
 543         function fieldInfo( $table, $field ) {
 544                 $res = $this->query( "SELECT $field FROM $table LIMIT 1" );
 545                 $type = pg_field_type( $res, 0 );
 546                 return $type;
 547         }
 548
 549         function begin( $fname = 'DatabasePostgrs::begin' ) {
 550                 $this->query( 'BEGIN', $fname );
 551                 $this->mTrxLevel = 1;
 552         }
 553         function immediateCommit( $fname = 'DatabasePostgres::immediateCommit' ) {
 554                 return true;
 555         }
 556         function commit( $fname = 'DatabasePostgres::commit' ) {
 557                 $this->query( 'COMMIT', $fname );
 558                 $this->mTrxLevel = 0;
 559         }
 560
 561         /* Not even sure why this is used in the main codebase... */
 562         function limitResultForUpdate($sql, $num) {
 563                 return $sql;
 564         }
 565
 566         function update_interwiki() {
 567                 ## Avoid the non-standard "REPLACE INTO" syntax
 568                 ## Called by config/index.php
 569                 $f = fopen( "../maintenance/interwiki.sql", 'r' );
 570                 if ($f == false ) {
 571                         dieout( "<li>Could not find the interwiki.sql file");
 572                 }
 573                 ## We simply assume it is already empty as we have just created it
 574                 $SQL = "INSERT INTO interwiki(iw_prefix,iw_url,iw_local) VALUES ";
 575                 while ( ! feof( $f ) ) {
 576                         $line = fgets($f,1024);
 577                         if (!preg_match("/^\s*(\(.+?),(\d)\)/", $line, $matches)) {
 578                                 continue;
 579                         }
 580                         $yesno = $matches[2]; ## ? "'true'" : "'false'";
 581                         $this->query("$SQL $matches[1],$matches[2])");
 582                 }
 583                 print " (table interwiki successfully populated)...\n";
 584         }
 585
 586         function encodeBlob($b) {
 587                 return array('bytea',pg_escape_bytea($b));
 588         }
 589         function decodeBlob($b) {
 590                 return pg_unescape_bytea( $b );
 591         }
 592
 593         function strencode( $s ) { ## Should not be called by us
 594                 return pg_escape_string( $s );
 595         }
 596
 597         function addQuotes( $s ) {
 598                 if ( is_null( $s ) ) {
 599                         return 'NULL';
 600                 } else if (is_array( $s )) { ## Assume it is bytea data
 601                         return "E'$s[1]'";
 602                 }
 603                 return "'" . pg_escape_string($s) . "'";
 604                 return "E'" . pg_escape_string($s) . "'";
 605         }
 606
 607 }
 608
 609 ?>