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 $wgDBschema, $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                 $schema = $wgDBschema;
  64
  65                 $success = false;
  66
  67                 $hstring="";
  68                 if ($server!=false && $server!="") {
  69                         $hstring="host=$server ";
  70                 }
  71                 if ($port!=false && $port!="") {
  72                         $hstring .= "port=$port ";
  73                 }
  74
  75                 error_reporting( E_ALL );
  76
  77                 @$this->mConn = pg_connect("$hstring dbname=$dbName user=$user password=$password");
  78
  79                 if ( $this->mConn == false ) {
  80                         wfDebug( "DB connection error\n" );
  81                         wfDebug( "Server: $server, Database: $dbName, User: $user, Password: " . substr( $password, 0, 3 ) . "...\n" );
  82                         wfDebug( $this->lastError()."\n" );
  83                         return false;
  84                 }
  85
  86                 $this->mOpened = true;
  87                 ## If this is the initial connection, setup the schema stuff
  88                 if (defined('MEDIAWIKI_INSTALL')) {
  89                         ## Does the schema already exist? Who owns it?
  90                         $result = $this->schemaExists($schema);
  91                         if (!$result) {
  92                                 print "<li>Creating schema <b>$schema</b> ...";
  93                                 $result = $this->doQuery("CREATE SCHEMA $schema");
  94                                 if (!$result) {
  95                                         print "FAILED.</li>\n";
  96                                         return false;
  97                                 }
  98                                 print "ok</li>\n";
  99                         }
 100                         else if ($result != $user) {
 101                                 print "<li>Schema <b>$schema</b> exists but is not owned by <b>$user</b>. Not ideal.</li>\n";
 102                         }
 103                         else {
 104                                 print "<li>Schema <b>$schema</b> exists and is owned by <b>$user ($result)</b>. Excellent.</li>\n";
 105                         }
 106
 107                         ## Fix up the search paths if needed
 108                         print "<li>Setting the search path for user <b>$user</b> ...";
 109                         $SQL = "ALTER USER $user SET search_path = $schema, public";
 110                         $result = pg_query($this->mConn, $SQL);
 111                         if (!$result) {
 112                                 print "FAILED.</li>\n";
 113                                 return false;
 114                         }
 115                         print "ok</li>\n";
 116                         ## Set for the rest of this session
 117                         $SQL = "SET search_path = $schema, public";
 118                         $result = pg_query($this->mConn, $SQL);
 119                         if (!$result) {
 120                                 print "<li>Failed to set search_path</li>\n";
 121                                 return false;
 122                         }
 123                 }
 124
 125                 return $this->mConn;
 126         }
 127
 128         /**
 129          * Closes a database connection, if it is open
 130          * Returns success, true if already closed
 131          */
 132         function close() {
 133                 $this->mOpened = false;
 134                 if ( $this->mConn ) {
 135                         return pg_close( $this->mConn );
 136                 } else {
 137                         return true;
 138                 }
 139         }
 140
 141         function doQuery( $sql ) {
 142                 return $this->mLastResult=pg_query( $this->mConn , $sql);
 143         }
 144
 145         function queryIgnore( $sql, $fname = '' ) {
 146                 return $this->query( $sql, $fname, true );
 147         }
 148
 149         function freeResult( $res ) {
 150                 if ( !@pg_free_result( $res ) ) {
 151                         throw new DBUnexpectedError($this,  "Unable to free PostgreSQL result\n" );
 152                 }
 153         }
 154
 155         function fetchObject( $res ) {
 156                 @$row = pg_fetch_object( $res );
 157                 # FIXME: HACK HACK HACK HACK debug
 158
 159                 # TODO:
 160                 # hashar : not sure if the following test really trigger if the object
 161                 #          fetching failled.
 162                 if( pg_last_error($this->mConn) ) {
 163                         throw new DBUnexpectedError($this,  'SQL error: ' . htmlspecialchars( pg_last_error($this->mConn) ) );
 164                 }
 165                 return $row;
 166         }
 167
 168         function fetchRow( $res ) {
 169                 @$row = pg_fetch_array( $res );
 170                 if( pg_last_error($this->mConn) ) {
 171                         throw new DBUnexpectedError($this,  'SQL error: ' . htmlspecialchars( pg_last_error($this->mConn) ) );
 172                 }
 173                 return $row;
 174         }
 175
 176         function numRows( $res ) {
 177                 @$n = pg_num_rows( $res );
 178                 if( pg_last_error($this->mConn) ) {
 179                         throw new DBUnexpectedError($this,  'SQL error: ' . htmlspecialchars( pg_last_error($this->mConn) ) );
 180                 }
 181                 return $n;
 182         }
 183         function numFields( $res ) { return pg_num_fields( $res ); }
 184         function fieldName( $res, $n ) { return pg_field_name( $res, $n ); }
 185
 186         /**
 187          * This must be called after nextSequenceVal
 188          */
 189         function insertId() {
 190                 return $this->mInsertId;
 191         }
 192
 193         function dataSeek( $res, $row ) { return pg_result_seek( $res, $row ); }
 194         function lastError() {
 195                 if ( $this->mConn ) {
 196                         return pg_last_error();
 197                 }
 198                 else {
 199                         return "No database connection";
 200                 }
 201         }
 202         function lastErrno() { return 1; }
 203
 204         function affectedRows() {
 205                 return pg_affected_rows( $this->mLastResult );
 206         }
 207
 208         /**
 209          * Returns information about an index
 210          * If errors are explicitly ignored, returns NULL on failure
 211          */
 212         function indexInfo( $table, $index, $fname = 'Database::indexExists' ) {
 213                 $sql = "SELECT indexname FROM pg_indexes WHERE tablename='$table'";
 214                 $res = $this->query( $sql, $fname );
 215                 if ( !$res ) {
 216                         return NULL;
 217                 }
 218
 219                 while ( $row = $this->fetchObject( $res ) ) {
 220                         if ( $row->indexname == $index ) {
 221                                 return $row;
 222                         }
 223                 }
 224                 return false;
 225         }
 226
 227         function indexUnique ($table, $index, $fname = 'Database::indexUnique' ) {
 228                 $sql = "SELECT indexname FROM pg_indexes WHERE tablename='{$table}'".
 229                         " AND indexdef LIKE 'CREATE UNIQUE%({$index})'";
 230                 $res = $this->query( $sql, $fname );
 231                 if ( !$res )
 232                         return NULL;
 233                 while ($row = $this->fetchObject( $res ))
 234                         return true;
 235                 return false;
 236
 237         }
 238
 239         function insert( $table, $a, $fname = 'Database::insert', $options = array() ) {
 240                 # PostgreSQL doesn't support options
 241                 # We have a go at faking one of them
 242                 # TODO: DELAYED, LOW_PRIORITY
 243
 244                 if ( !is_array($options))
 245                         $options = array($options);
 246
 247                 if ( in_array( 'IGNORE', $options ) )
 248                         $oldIgnore = $this->ignoreErrors( true );
 249
 250                 # IGNORE is performed using single-row inserts, ignoring errors in each
 251                 # FIXME: need some way to distiguish between key collision and other types of error
 252                 $oldIgnore = $this->ignoreErrors( true );
 253                 if ( !is_array( reset( $a ) ) ) {
 254                         $a = array( $a );
 255                 }
 256                 foreach ( $a as $row ) {
 257                         parent::insert( $table, $row, $fname, array() );
 258                 }
 259                 $this->ignoreErrors( $oldIgnore );
 260                 $retVal = true;
 261
 262                 if ( in_array( 'IGNORE', $options ) )
 263                         $this->ignoreErrors( $oldIgnore );
 264
 265                 return $retVal;
 266         }
 267
 268         function tableName( $name ) {
 269                 # Replace backticks into double quotes
 270                 $name = strtr($name,'`','"');
 271
 272                 # Now quote PG reserved keywords
 273                 switch( $name ) {
 274                         case 'user':
 275                         case 'old':
 276                         case 'group':
 277                                 return '"' . $name . '"';
 278
 279                         default:
 280                                 return $name;
 281                 }
 282         }
 283
 284         function strencode( $s ) {
 285                 return pg_escape_string( $s );
 286         }
 287
 288         /**
 289          * Return the next in a sequence, save the value for retrieval via insertId()
 290          */
 291         function nextSequenceValue( $seqName ) {
 292                 $value = $this->selectField(''," nextval('" . $seqName . "')");
 293                 $this->mInsertId = $value;
 294                 return $value;
 295         }
 296
 297         /**
 298          * USE INDEX clause
 299          * PostgreSQL doesn't have them and returns ""
 300          */
 301         function useIndexClause( $index ) {
 302                 return '';
 303         }
 304
 305         # REPLACE query wrapper
 306         # PostgreSQL simulates this with a DELETE followed by INSERT
 307         # $row is the row to insert, an associative array
 308         # $uniqueIndexes is an array of indexes. Each element may be either a
 309         # field name or an array of field names
 310         #
 311         # It may be more efficient to leave off unique indexes which are unlikely to collide.
 312         # However if you do this, you run the risk of encountering errors which wouldn't have
 313         # occurred in MySQL
 314         function replace( $table, $uniqueIndexes, $rows, $fname = 'Database::replace' ) {
 315                 $table = $this->tableName( $table );
 316
 317                 if (count($rows)==0) {
 318                         return;
 319                 }
 320
 321                 # Single row case
 322                 if ( !is_array( reset( $rows ) ) ) {
 323                         $rows = array( $rows );
 324                 }
 325
 326                 foreach( $rows as $row ) {
 327                         # Delete rows which collide
 328                         if ( $uniqueIndexes ) {
 329                                 $sql = "DELETE FROM $table WHERE ";
 330                                 $first = true;
 331                                 foreach ( $uniqueIndexes as $index ) {
 332                                         if ( $first ) {
 333                                                 $first = false;
 334                                                 $sql .= "(";
 335                                         } else {
 336                                                 $sql .= ') OR (';
 337                                         }
 338                                         if ( is_array( $index ) ) {
 339                                                 $first2 = true;
 340                                                 foreach ( $index as $col ) {
 341                                                         if ( $first2 ) {
 342                                                                 $first2 = false;
 343                                                         } else {
 344                                                                 $sql .= ' AND ';
 345                                                         }
 346                                                         $sql .= $col.'=' . $this->addQuotes( $row[$col] );
 347                                                 }
 348                                         } else {
 349                                                 $sql .= $index.'=' . $this->addQuotes( $row[$index] );
 350                                         }
 351                                 }
 352                                 $sql .= ')';
 353                                 $this->query( $sql, $fname );
 354                         }
 355
 356                         # Now insert the row
 357                         $sql = "INSERT INTO $table (" . $this->makeList( array_keys( $row ), LIST_NAMES ) .') VALUES (' .
 358                                 $this->makeList( $row, LIST_COMMA ) . ')';
 359                         $this->query( $sql, $fname );
 360                 }
 361         }
 362
 363         # DELETE where the condition is a join
 364         function deleteJoin( $delTable, $joinTable, $delVar, $joinVar, $conds, $fname = "Database::deleteJoin" ) {
 365                 if ( !$conds ) {
 366                         throw new DBUnexpectedError($this,  'Database::deleteJoin() called with empty $conds' );
 367                 }
 368
 369                 $delTable = $this->tableName( $delTable );
 370                 $joinTable = $this->tableName( $joinTable );
 371                 $sql = "DELETE FROM $delTable WHERE $delVar IN (SELECT $joinVar FROM $joinTable ";
 372                 if ( $conds != '*' ) {
 373                         $sql .= 'WHERE ' . $this->makeList( $conds, LIST_AND );
 374                 }
 375                 $sql .= ')';
 376
 377                 $this->query( $sql, $fname );
 378         }
 379
 380         # Returns the size of a text field, or -1 for "unlimited"
 381         function textFieldSize( $table, $field ) {
 382                 $table = $this->tableName( $table );
 383                 $sql = "SELECT t.typname as ftype,a.atttypmod as size
 384                         FROM pg_class c, pg_attribute a, pg_type t
 385                         WHERE relname='$table' AND a.attrelid=c.oid AND
 386                                 a.atttypid=t.oid and a.attname='$field'";
 387                 $res =$this->query($sql);
 388                 $row=$this->fetchObject($res);
 389                 if ($row->ftype=="varchar") {
 390                         $size=$row->size-4;
 391                 } else {
 392                         $size=$row->size;
 393                 }
 394                 $this->freeResult( $res );
 395                 return $size;
 396         }
 397
 398         function lowPriorityOption() {
 399                 return '';
 400         }
 401
 402         function limitResult($sql, $limit,$offset) {
 403                 return "$sql LIMIT $limit ".(is_numeric($offset)?" OFFSET {$offset} ":"");
 404         }
 405
 406         /**
 407          * Returns an SQL expression for a simple conditional.
 408          * Uses CASE on PostgreSQL.
 409          *
 410          * @param string $cond SQL expression which will result in a boolean value
 411          * @param string $trueVal SQL expression to return if true
 412          * @param string $falseVal SQL expression to return if false
 413          * @return string SQL fragment
 414          */
 415         function conditional( $cond, $trueVal, $falseVal ) {
 416                 return " (CASE WHEN $cond THEN $trueVal ELSE $falseVal END) ";
 417         }
 418
 419         # FIXME: actually detecting deadlocks might be nice
 420         function wasDeadlock() {
 421                 return false;
 422         }
 423
 424         # Return DB-style timestamp used for MySQL schema
 425         function timestamp( $ts=0 ) {
 426                 return wfTimestamp(TS_DB,$ts);
 427         }
 428
 429         /**
 430          * Return aggregated value function call
 431          */
 432         function aggregateValue ($valuedata,$valuename='value') {
 433                 return $valuedata;
 434         }
 435
 436
 437         function reportQueryError( $error, $errno, $sql, $fname, $tempIgnore = false ) {
 438                 $message = "A database error has occurred\n" .
 439                         "Query: $sql\n" .
 440                         "Function: $fname\n" .
 441                         "Error: $errno $error\n";
 442                 throw new DBUnexpectedError($this, $message);
 443         }
 444
 445         /**
 446          * @return string wikitext of a link to the server software's web site
 447          */
 448         function getSoftwareLink() {
 449                 return "[http://www.postgresql.org/ PostgreSQL]";
 450         }
 451
 452         /**
 453          * @return string Version information from the database
 454          */
 455         function getServerVersion() {
 456                 $res = $this->query( "SELECT version()" );
 457                 $row = $this->fetchRow( $res );
 458                 $version = $row[0];
 459                 $this->freeResult( $res );
 460                 return $version;
 461         }
 462
 463
 464         /**
 465          * Query whether a given table exists (in the default schema)
 466          */
 467         function tableExists( $table, $fname = 'DatabasePostgres:tableExists' ) {
 468                 global $wgDBschema;
 469                 $stable = preg_replace("/'/", "''", $table);
 470                 $SQL = "SELECT 1 FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n "
 471                         . "WHERE c.relnamespace = n.oid AND c.relname = '$stable' AND n.nspname = '$wgDBschema'";
 472                 $res = $this->query( $SQL, $fname );
 473                 $count = $res ? pg_num_rows($res) : 0;
 474                 if ($res)
 475                         $this->freeResult( $res );
 476                 return $count;
 477         }
 478
 479         /**
 480          * Query whether a given schema exists. Returns the name of the owner
 481          */
 482         function schemaExists( $schema, $fname = 'DatabasePostgres:schemaExists' ) {
 483                 $sschema = preg_replace("/'/", "''", $schema);
 484                 $SQL = "SELECT rolname FROM pg_catalog.pg_namespace n, pg_catalog.pg_roles r "
 485                                 ."WHERE n.nspowner=r.oid AND n.nspname = '$sschema'";
 486                 $res = $this->query($SQL, $fname);
 487                 $res = $this->query( $SQL, $fname );
 488                 $owner = $res ? pg_num_rows($res) ? pg_fetch_result($res, 0, 0) : false : false;
 489                 if ($res)
 490                         $this->freeResult($res);
 491                 return $owner;
 492         }
 493
 494         /**
 495          * Query whether a given column exists
 496          */
 497         function fieldExists( $table, $field, $fname = 'DatabasePostgres::fieldExists' ) {
 498                 global $wgDBschema;
 499                 $stable = preg_replace("/'/", "''", $table);
 500                 $scol = preg_replace("/'/", "''", $field);
 501                 $SQL = "SELECT 1 FROM pg_catalog.pg_class c, pg_catalog.pg_namespace n, pg_catalog.pg_attribute a "
 502                         . "WHERE c.relnamespace = n.oid AND c.relname = '$stable' AND n.nspname = '$wgDBschema' "
 503                         . "AND a.attrelid = c.oid AND a.attname = '$scol'";
 504                 $res = $this->query( $SQL, $fname );
 505                 $count = $res ? pg_num_rows($res) : 0;
 506                 if ($res)
 507                         $this->freeResult( $res );
 508                 return $count;
 509         }
 510
 511         function fieldInfo( $table, $field ) {
 512                 $res = $this->query( "SELECT $field FROM $table LIMIT 1" );
 513                 $type = pg_field_type( $res, 0 );
 514                 return $type;
 515         }
 516
 517 }
 518
 519 ?>