includes/db/DatabaseMssql.php

   1 <?php
   2 /**
   3  * This is the MS SQL Server Native database abstraction layer.
   4  *
   5  * @file
   6  * @ingroup Database
   7  * @author Joel Penner <a-joelpe at microsoft dot com>
   8  * @author Chris Pucci <a-cpucci at microsoft dot com>
   9  * @author Ryan Biesemeyer <v-ryanbi at microsoft dot com>
  10  */
  11
  12 /**
  13  * @ingroup Database
  14  */
  15 class DatabaseMssql extends DatabaseBase {
  16         var $mInsertId = NULL;
  17         var $mLastResult = NULL;
  18         var $mAffectedRows = NULL;
  19
  20         var $mPort;
  21
  22         function cascadingDeletes() {
  23                 return true;
  24         }
  25         function cleanupTriggers() {
  26                 return true;
  27         }
  28         function strictIPs() {
  29                 return true;
  30         }
  31         function realTimestamps() {
  32                 return true;
  33         }
  34         function implicitGroupby() {
  35                 return false;
  36         }
  37         function implicitOrderby() {
  38                 return false;
  39         }
  40         function functionalIndexes() {
  41                 return true;
  42         }
  43         function unionSupportsOrderAndLimit() {
  44                 return false;
  45         }
  46
  47         /**
  48          * Usually aborts on failure
  49          */
  50         function open( $server, $user, $password, $dbName ) {
  51                 # Test for driver support, to avoid suppressed fatal error
  52                 if ( !function_exists( 'sqlsrv_connect' ) ) {
  53                         throw new DBConnectionError( $this, "MS Sql Server Native (sqlsrv) functions missing. You can download the driver from: http://go.microsoft.com/fwlink/?LinkId=123470\n" );
  54                 }
  55
  56                 global $wgDBport;
  57
  58                 if ( !strlen( $user ) ) { # e.g. the class is being loaded
  59                         return;
  60                 }
  61
  62                 $this->close();
  63                 $this->mServer = $server;
  64                 $this->mPort = $wgDBport;
  65                 $this->mUser = $user;
  66                 $this->mPassword = $password;
  67                 $this->mDBname = $dbName;
  68
  69                 $connectionInfo = array();
  70
  71                 if( $dbName ) {
  72                         $connectionInfo['Database'] = $dbName;
  73                 }
  74
  75                 // Start NT Auth Hack
  76                 // Quick and dirty work around to provide NT Auth designation support.
  77                 // Current solution requires installer to know to input 'ntauth' for both username and password
  78                 // to trigger connection via NT Auth. - ugly, ugly, ugly
  79                 // TO-DO: Make this better and add NT Auth choice to MW installer when SQL Server option is chosen.
  80                 $ntAuthUserTest = strtolower( $user );
  81                 $ntAuthPassTest = strtolower( $password );
  82
  83                 // Decide which auth scenerio to use
  84                 if( $ntAuthPassTest == 'ntauth' && $ntAuthUserTest == 'ntauth' ){
  85                         // Don't add credentials to $connectionInfo
  86                 } else {
  87                         $connectionInfo['UID'] = $user;
  88                         $connectionInfo['PWD'] = $password;
  89                 }
  90                 // End NT Auth Hack
  91
  92                 wfSuppressWarnings();
  93                 $this->mConn = sqlsrv_connect( $server, $connectionInfo );
  94                 wfRestoreWarnings();
  95
  96                 if ( $this->mConn === false ) {
  97                         wfDebug( "DB connection error\n" );
  98                         wfDebug( "Server: $server, Database: $dbName, User: $user, Password: " . substr( $password, 0, 3 ) . "...\n" );
  99                         wfDebug( $this->lastError() . "\n" );
 100                         return false;
 101                 }
 102
 103                 $this->mOpened = true;
 104                 return $this->mConn;
 105         }
 106
 107         /**
 108          * Closes a database connection, if it is open
 109          * Returns success, true if already closed
 110          */
 111         function close() {
 112                 $this->mOpened = false;
 113                 if ( $this->mConn ) {
 114                         return sqlsrv_close( $this->mConn );
 115                 } else {
 116                         return true;
 117                 }
 118         }
 119
 120         protected function doQuery( $sql ) {
 121                 wfDebug( "SQL: [$sql]\n" );
 122                 $this->offset = 0;
 123
 124                 // several extensions seem to think that all databases support limits via LIMIT N after the WHERE clause
 125                 // well, MSSQL uses SELECT TOP N, so to catch any of those extensions we'll do a quick check for a LIMIT
 126                 // clause and pass $sql through $this->LimitToTopN() which parses the limit clause and passes the result to
 127                 // $this->limitResult();
 128                 if ( preg_match( '/\bLIMIT\s*/i', $sql ) ) {
 129                         // massage LIMIT -> TopN
 130                         $sql = $this->LimitToTopN( $sql ) ;
 131                 }
 132
 133                 // MSSQL doesn't have EXTRACT(epoch FROM XXX)
 134                 if ( preg_match('#\bEXTRACT\s*?\(\s*?EPOCH\s+FROM\b#i', $sql, $matches ) ) {
 135                         // This is same as UNIX_TIMESTAMP, we need to calc # of seconds from 1970
 136                         $sql = str_replace( $matches[0], "DATEDIFF(s,CONVERT(datetime,'1/1/1970'),", $sql );
 137                 }
 138
 139                 // perform query
 140                 $stmt = sqlsrv_query( $this->mConn, $sql );
 141                 if ( $stmt == false ) {
 142                         $message = "A database error has occurred.  Did you forget to run maintenance/update.php after upgrading?  See: http://www.mediawiki.org/wiki/Manual:Upgrading#Run_the_update_script\n" .
 143                                 "Query: " . htmlentities( $sql ) . "\n" .
 144                                 "Function: " . __METHOD__ . "\n";
 145                         // process each error (our driver will give us an array of errors unlike other providers)
 146                         foreach ( sqlsrv_errors() as $error ) {
 147                                 $message .= $message . "ERROR[" . $error['code'] . "] " . $error['message'] . "\n";
 148                         }
 149
 150                         throw new DBUnexpectedError( $this, $message );
 151                 }
 152                 // remember number of rows affected
 153                 $this->mAffectedRows = sqlsrv_rows_affected( $stmt );
 154
 155                 // if it is a SELECT statement, or an insert with a request to output something we want to return a row.
 156                 if ( ( preg_match( '#\bSELECT\s#i', $sql ) ) ||
 157                         ( preg_match( '#\bINSERT\s#i', $sql ) && preg_match( '#\bOUTPUT\s+INSERTED\b#i', $sql ) ) ) {
 158                         // this is essentially a rowset, but Mediawiki calls these 'result'
 159                         // the rowset owns freeing the statement
 160                         $res = new MssqlResult( $stmt );
 161                 } else {
 162                         // otherwise we simply return it was successful, failure throws an exception
 163                         $res = true;
 164                 }
 165                 return $res;
 166         }
 167
 168         function freeResult( $res ) {
 169                 if ( $res instanceof ResultWrapper ) {
 170                         $res = $res->result;
 171                 }
 172                 $res->free();
 173         }
 174
 175         function fetchObject( $res ) {
 176                 if ( $res instanceof ResultWrapper ) {
 177                         $res = $res->result;
 178                 }
 179                 $row = $res->fetch( 'OBJECT' );
 180                 return $row;
 181         }
 182
 183         function getErrors() {
 184                 $strRet = '';
 185                 $retErrors = sqlsrv_errors( SQLSRV_ERR_ALL );
 186                 if ( $retErrors != null ) {
 187                         foreach ( $retErrors as $arrError ) {
 188                                 $strRet .= "SQLState: " . $arrError[ 'SQLSTATE'] . "\n";
 189                                 $strRet .= "Error Code: " . $arrError[ 'code'] . "\n";
 190                                 $strRet .= "Message: " . $arrError[ 'message'] . "\n";
 191                         }
 192                 } else {
 193                         $strRet = "No errors found";
 194                 }
 195                 return $strRet;
 196         }
 197
 198         function fetchRow( $res ) {
 199                 if ( $res instanceof ResultWrapper ) {
 200                         $res = $res->result;
 201                 }
 202                 $row = $res->fetch( SQLSRV_FETCH_BOTH );
 203                 return $row;
 204         }
 205
 206         function numRows( $res ) {
 207                 if ( $res instanceof ResultWrapper ) {
 208                         $res = $res->result;
 209                 }
 210                 return ( $res ) ? $res->numrows() : 0;
 211         }
 212
 213         function numFields( $res ) {
 214                 if ( $res instanceof ResultWrapper ) {
 215                         $res = $res->result;
 216                 }
 217                 return ( $res ) ? $res->numfields() : 0;
 218         }
 219
 220         function fieldName( $res, $n ) {
 221                 if ( $res instanceof ResultWrapper ) {
 222                         $res = $res->result;
 223                 }
 224                 return ( $res ) ? $res->fieldname( $n ) : 0;
 225         }
 226
 227         /**
 228          * This must be called after nextSequenceVal
 229          */
 230         function insertId() {
 231                 return $this->mInsertId;
 232         }
 233
 234         function dataSeek( $res, $row ) {
 235                 if ( $res instanceof ResultWrapper ) {
 236                         $res = $res->result;
 237                 }
 238                 return ( $res ) ? $res->seek( $row ) : false;
 239         }
 240
 241         function lastError() {
 242                 if ( $this->mConn ) {
 243                         return $this->getErrors();
 244                 } else {
 245                         return "No database connection";
 246                 }
 247         }
 248
 249         function lastErrno() {
 250                 $err = sqlsrv_errors( SQLSRV_ERR_ALL );
 251                 if ( $err[0] ) {
 252                         return $err[0]['code'];
 253                 } else {
 254                         return 0;
 255                 }
 256         }
 257
 258         function affectedRows() {
 259                 return $this->mAffectedRows;
 260         }
 261
 262         /**
 263          * SELECT wrapper
 264          *
 265          * @param $table   Mixed: array or string, table name(s) (prefix auto-added)
 266          * @param $vars    Mixed: array or string, field name(s) to be retrieved
 267          * @param $conds   Mixed: array or string, condition(s) for WHERE
 268          * @param $fname   String: calling function name (use __METHOD__) for logs/profiling
 269          * @param $options Array: associative array of options (e.g. array('GROUP BY' => 'page_title')),
 270          *                 see Database::makeSelectOptions code for list of supported stuff
 271          * @param $join_conds Array: Associative array of table join conditions (optional)
 272          *                                                 (e.g. array( 'page' => array('LEFT JOIN','page_latest=rev_id') )
 273          * @return Mixed: database result resource (feed to Database::fetchObject or whatever), or false on failure
 274          */
 275         function select( $table, $vars, $conds = '', $fname = 'DatabaseMssql::select', $options = array(), $join_conds = array() )
 276         {
 277                 $sql = $this->selectSQLText( $table, $vars, $conds, $fname, $options, $join_conds );
 278                 if ( isset( $options['EXPLAIN'] ) ) {
 279                         sqlsrv_query( $this->mConn, "SET SHOWPLAN_ALL ON;" );
 280                         $ret = $this->query( $sql, $fname );
 281                         sqlsrv_query( $this->mConn, "SET SHOWPLAN_ALL OFF;" );
 282                         return $ret;
 283                 }
 284                 return $this->query( $sql, $fname );
 285         }
 286
 287         /**
 288          * SELECT wrapper
 289          *
 290          * @param $table   Mixed:  Array or string, table name(s) (prefix auto-added)
 291          * @param $vars    Mixed:  Array or string, field name(s) to be retrieved
 292          * @param $conds   Mixed:  Array or string, condition(s) for WHERE
 293          * @param $fname   String: Calling function name (use __METHOD__) for logs/profiling
 294          * @param $options Array:  Associative array of options (e.g. array('GROUP BY' => 'page_title')),
 295          *                 see Database::makeSelectOptions code for list of supported stuff
 296          * @param $join_conds Array: Associative array of table join conditions (optional)
 297          *                    (e.g. array( 'page' => array('LEFT JOIN','page_latest=rev_id') )
 298          * @return string, the SQL text
 299          */
 300         function selectSQLText( $table, $vars, $conds = '', $fname = 'DatabaseMssql::select', $options = array(), $join_conds = array() ) {
 301                 if ( isset( $options['EXPLAIN'] ) ) {
 302                         unset( $options['EXPLAIN'] );
 303                 }
 304                 return parent::selectSQLText(  $table, $vars, $conds, $fname, $options, $join_conds );
 305         }
 306
 307         /**
 308          * Estimate rows in dataset
 309          * Returns estimated count, based on SHOWPLAN_ALL output
 310          * This is not necessarily an accurate estimate, so use sparingly
 311          * Returns -1 if count cannot be found
 312          * Takes same arguments as Database::select()
 313          */
 314         function estimateRowCount( $table, $vars = '*', $conds = '', $fname = 'DatabaseMssql::estimateRowCount', $options = array() ) {
 315                 $options['EXPLAIN'] = true;// http://msdn2.microsoft.com/en-us/library/aa259203.aspx
 316                 $res = $this->select( $table, $vars, $conds, $fname, $options );
 317
 318                 $rows = -1;
 319                 if ( $res ) {
 320                         $row = $this->fetchRow( $res );
 321                         if ( isset( $row['EstimateRows'] ) ) $rows = $row['EstimateRows'];
 322                 }
 323                 return $rows;
 324         }
 325
 326         /**
 327          * Returns information about an index
 328          * If errors are explicitly ignored, returns NULL on failure
 329          */
 330         function indexInfo( $table, $index, $fname = 'DatabaseMssql::indexExists' ) {
 331                 # This does not return the same info as MYSQL would, but that's OK because MediaWiki never uses the
 332                 # returned value except to check for the existance of indexes.
 333                 $sql = "sp_helpindex '" . $table . "'";
 334                 $res = $this->query( $sql, $fname );
 335                 if ( !$res ) {
 336                         return NULL;
 337                 }
 338
 339                 $result = array();
 340                 foreach ( $res as $row ) {
 341                         if ( $row->index_name == $index ) {
 342                                 $row->Non_unique = !stristr( $row->index_description, "unique" );
 343                                 $cols = explode( ", ", $row->index_keys );
 344                                 foreach ( $cols as $col ) {
 345                                         $row->Column_name = trim( $col );
 346                                         $result[] = clone $row;
 347                                 }
 348                         } elseif ( $index == 'PRIMARY' && stristr( $row->index_description, 'PRIMARY' ) ) {
 349                                 $row->Non_unique = 0;
 350                                 $cols = explode( ", ", $row->index_keys );
 351                                 foreach ( $cols as $col ) {
 352                                         $row->Column_name = trim( $col );
 353                                         $result[] = clone $row;
 354                                 }
 355                         }
 356                 }
 357                 return empty( $result ) ? false : $result;
 358         }
 359
 360         /**
 361          * INSERT wrapper, inserts an array into a table
 362          *
 363          * $arrToInsert may be a single associative array, or an array of these with numeric keys, for
 364          * multi-row insert.
 365          *
 366          * Usually aborts on failure
 367          * If errors are explicitly ignored, returns success
 368          */
 369         function insert( $table, $arrToInsert, $fname = 'DatabaseMssql::insert', $options = array() ) {
 370                 # No rows to insert, easy just return now
 371                 if ( !count( $arrToInsert ) ) {
 372                         return true;
 373                 }
 374
 375                 if ( !is_array( $options ) ) {
 376                         $options = array( $options );
 377                 }
 378
 379                 $table = $this->tableName( $table );
 380
 381                 if ( !( isset( $arrToInsert[0] ) && is_array( $arrToInsert[0] ) ) ) {// Not multi row
 382                         $arrToInsert = array( 0 => $arrToInsert );// make everything multi row compatible
 383                 }
 384
 385                 $allOk = true;
 386
 387                 // We know the table we're inserting into, get its identity column
 388                 $identity = null;
 389                 $tableRaw = preg_replace( '#\[([^\]]*)\]#', '$1', $table ); // strip matching square brackets from table name
 390                 $res = $this->doQuery( "SELECT NAME AS idColumn FROM SYS.IDENTITY_COLUMNS WHERE OBJECT_NAME(OBJECT_ID)='{$tableRaw}'" );
 391                 if( $res && $res->numrows() ){
 392                         // There is an identity for this table.
 393                         $identity = array_pop( $res->fetch( SQLSRV_FETCH_ASSOC ) );
 394                 }
 395                 unset( $res );
 396
 397                 foreach ( $arrToInsert as $a ) {
 398                         // start out with empty identity column, this is so we can return it as a result of the insert logic
 399                         $sqlPre = '';
 400                         $sqlPost = '';
 401                         $identityClause = '';
 402
 403                         // if we have an identity column
 404                         if( $identity ) {
 405                                 // iterate through
 406                                 foreach ($a as $k => $v ) {
 407                                         if ( $k == $identity ) {
 408                                                 if( !is_null($v) ){
 409                                                         // there is a value being passed to us, we need to turn on and off inserted identity
 410                                                         $sqlPre = "SET IDENTITY_INSERT $table ON;" ;
 411                                                         $sqlPost = ";SET IDENTITY_INSERT $table OFF;";
 412
 413                                                 } else {
 414                                                         // we can't insert NULL into an identity column, so remove the column from the insert.
 415                                                         unset( $a[$k] );
 416                                                 }
 417                                         }
 418                                 }
 419                                 $identityClause = "OUTPUT INSERTED.$identity "; // we want to output an identity column as result
 420                         }
 421
 422                         $keys = array_keys( $a );
 423
 424                         // INSERT IGNORE is not supported by SQL Server
 425                         // remove IGNORE from options list and set ignore flag to true
 426                         $ignoreClause = false;
 427                         foreach ( $options as $k => $v ) {
 428                                 if ( strtoupper( $v ) == "IGNORE" ) {
 429                                         unset( $options[$k] );
 430                                         $ignoreClause = true;
 431                                 }
 432                         }
 433
 434                         // translate MySQL INSERT IGNORE to something SQL Server can use
 435                         // example:
 436                         // MySQL: INSERT IGNORE INTO user_groups (ug_user,ug_group) VALUES ('1','sysop')
 437                         // MSSQL: IF NOT EXISTS (SELECT * FROM user_groups WHERE ug_user = '1') INSERT INTO user_groups (ug_user,ug_group) VALUES ('1','sysop')
 438                         if ( $ignoreClause ) {
 439                                 $prival = $a[$keys[0]];
 440                                 $sqlPre .= "IF NOT EXISTS (SELECT * FROM $table WHERE $keys[0] = '$prival')";
 441                         }
 442
 443                         // Build the actual query
 444                         $sql = $sqlPre . 'INSERT ' . implode( ' ', $options ) .
 445                                 " INTO $table (" . implode( ',', $keys ) . ") $identityClause VALUES (";
 446
 447                         $first = true;
 448                         foreach ( $a as $value ) {
 449                                 if ( $first ) {
 450                                         $first = false;
 451                                 } else {
 452                                         $sql .= ',';
 453                                 }
 454                                 if ( is_string( $value ) ) {
 455                                         $sql .= $this->addQuotes( $value );
 456                                 } elseif ( is_null( $value ) ) {
 457                                         $sql .= 'null';
 458                                 } elseif ( is_array( $value ) || is_object( $value ) ) {
 459                                         if ( is_object( $value ) && strtolower( get_class( $value ) ) == 'blob' ) {
 460                                                 $sql .= $this->addQuotes( $value );
 461                                         }  else {
 462                                                 $sql .= $this->addQuotes( serialize( $value ) );
 463                                         }
 464                                 } else {
 465                                         $sql .= $value;
 466                                 }
 467                         }
 468                         $sql .= ')' . $sqlPost;
 469
 470                         // Run the query
 471                         $ret = sqlsrv_query( $this->mConn, $sql );
 472
 473                         if ( $ret === false ) {
 474                                 throw new DBQueryError( $this, $this->getErrors(), $this->lastErrno(), $sql, $fname );
 475                         } elseif ( $ret != NULL ) {
 476                                 // remember number of rows affected
 477                                 $this->mAffectedRows = sqlsrv_rows_affected( $ret );
 478                                 if ( !is_null($identity) ) {
 479                                         // then we want to get the identity column value we were assigned and save it off
 480                                         $row = sqlsrv_fetch_object( $ret );
 481                                         $this->mInsertId = $row->$identity;
 482                                 }
 483                                 sqlsrv_free_stmt( $ret );
 484                                 continue;
 485                         }
 486                         $allOk = false;
 487                 }
 488                 return $allOk;
 489         }
 490
 491         /**
 492          * INSERT SELECT wrapper
 493          * $varMap must be an associative array of the form array( 'dest1' => 'source1', ...)
 494          * Source items may be literals rather than field names, but strings should be quoted with Database::addQuotes()
 495          * $conds may be "*" to copy the whole table
 496          * srcTable may be an array of tables.
 497          */
 498         function insertSelect( $destTable, $srcTable, $varMap, $conds, $fname = 'DatabaseMssql::insertSelect',
 499                 $insertOptions = array(), $selectOptions = array() ) {
 500                 $ret = parent::insertSelect( $destTable, $srcTable, $varMap, $conds, $fname, $insertOptions, $selectOptions );
 501
 502                 if ( $ret === false ) {
 503                         throw new DBQueryError( $this, $this->getErrors(), $this->lastErrno(), /*$sql*/ '', $fname );
 504                 } elseif ( $ret != NULL ) {
 505                         // remember number of rows affected
 506                         $this->mAffectedRows = sqlsrv_rows_affected( $ret );
 507                         return $ret;
 508                 }
 509                 return NULL;
 510         }
 511
 512         /**
 513          * Return the next in a sequence, save the value for retrieval via insertId()
 514          */
 515         function nextSequenceValue( $seqName ) {
 516                 if ( !$this->tableExists( 'sequence_' . $seqName ) ) {
 517                         sqlsrv_query( $this->mConn, "CREATE TABLE [sequence_$seqName] (id INT NOT NULL IDENTITY PRIMARY KEY, junk varchar(10) NULL)" );
 518                 }
 519                 sqlsrv_query( $this->mConn, "INSERT INTO [sequence_$seqName] (junk) VALUES ('')" );
 520                 $ret = sqlsrv_query( $this->mConn, "SELECT TOP 1 id FROM [sequence_$seqName] ORDER BY id DESC" );
 521                 $row = sqlsrv_fetch_array( $ret, SQLSRV_FETCH_ASSOC );// KEEP ASSOC THERE, weird weird bug dealing with the return value if you don't
 522
 523                 sqlsrv_free_stmt( $ret );
 524                 $this->mInsertId = $row['id'];
 525                 return $row['id'];
 526         }
 527
 528         /**
 529          * Return the current value of a sequence. Assumes it has ben nextval'ed in this session.
 530          */
 531         function currentSequenceValue( $seqName ) {
 532                 $ret = sqlsrv_query( $this->mConn, "SELECT TOP 1 id FROM [sequence_$seqName] ORDER BY id DESC" );
 533                 if ( $ret !== false ) {
 534                         $row = sqlsrv_fetch_array( $ret );
 535                         sqlsrv_free_stmt( $ret );
 536                         return $row['id'];
 537                 } else {
 538                         return $this->nextSequenceValue( $seqName );
 539                 }
 540         }
 541
 542         # Returns the size of a text field, or -1 for "unlimited"
 543         function textFieldSize( $table, $field ) {
 544                 $table = $this->tableName( $table );
 545                 $sql = "SELECT CHARACTER_MAXIMUM_LENGTH,DATA_TYPE FROM INFORMATION_SCHEMA.Columns
 546                         WHERE TABLE_NAME = '$table' AND COLUMN_NAME = '$field'";
 547                 $res = $this->query( $sql );
 548                 $row = $this->fetchRow( $res );
 549                 $size = -1;
 550                 if ( strtolower( $row['DATA_TYPE'] ) != 'text' ) {
 551                         $size = $row['CHARACTER_MAXIMUM_LENGTH'];
 552                 }
 553                 return $size;
 554         }
 555
 556         /**
 557          * Construct a LIMIT query with optional offset
 558          * This is used for query pages
 559          * $sql string SQL query we will append the limit too
 560          * $limit integer the SQL limit
 561          * $offset integer the SQL offset (default false)
 562          */
 563         function limitResult( $sql, $limit, $offset = false ) {
 564                 if ( $offset === false || $offset == 0 ) {
 565                         if ( strpos( $sql, "SELECT" ) === false ) {
 566                                 return "TOP {$limit} " . $sql;
 567                         } else {
 568                                 return preg_replace( '/\bSELECT(\s*DISTINCT)?\b/Dsi', 'SELECT$1 TOP ' . $limit, $sql, 1 );
 569                         }
 570                 } else {
 571                         $sql = '
 572                                 SELECT * FROM (
 573                                   SELECT sub2.*, ROW_NUMBER() OVER(ORDER BY sub2.line2) AS line3 FROM (
 574                                         SELECT 1 AS line2, sub1.* FROM (' . $sql . ') AS sub1
 575                                   ) as sub2
 576                                 ) AS sub3
 577                                 WHERE line3 BETWEEN ' . ( $offset + 1 ) . ' AND ' . ( $offset + $limit );
 578                         return $sql;
 579                 }
 580         }
 581
 582         // If there is a limit clause, parse it, strip it, and pass the remaining sql through limitResult()
 583         // with the appropriate parameters. Not the prettiest solution, but better than building a whole new parser.
 584         // This exists becase there are still too many extensions that don't use dynamic sql generation.
 585         function LimitToTopN( $sql ) {
 586                 // Matches: LIMIT {[offset,] row_count | row_count OFFSET offset}
 587                 $pattern = '/\bLIMIT\s+((([0-9]+)\s*,\s*)?([0-9]+)(\s+OFFSET\s+([0-9]+))?)/i';
 588                 if ( preg_match( $pattern, $sql, $matches ) ) {
 589                         // row_count = $matches[4]
 590                         $row_count = $matches[4];
 591                         // offset = $matches[3] OR $matches[6]
 592                         $offset = $matches[3] or
 593                                 $offset = $matches[6] or
 594                                 $offset = false;
 595
 596                         // strip the matching LIMIT clause out
 597                         $sql = str_replace( $matches[0], '', $sql );
 598                         return $this->limitResult( $sql, $row_count, $offset );
 599                 }
 600                 return $sql;
 601         }
 602
 603         // MSSQL does support this, but documentation is too thin to make a generalized
 604         // function for this. Apparently UPDATE TOP (N) works, but the sort order
 605         // may not be what we're expecting so the top n results may be a random selection.
 606         // TODO: Implement properly.
 607         function limitResultForUpdate( $sql, $num ) {
 608                 return $sql;
 609         }
 610
 611         function timestamp( $ts = 0 ) {
 612                 return wfTimestamp( TS_ISO_8601, $ts );
 613         }
 614
 615         /**
 616          * @return string wikitext of a link to the server software's web site
 617          */
 618         public static function getSoftwareLink() {
 619                 return "[http://www.microsoft.com/sql/ MS SQL Server]";
 620         }
 621
 622         /**
 623          * @return string Version information from the database
 624          */
 625         function getServerVersion() {
 626                 $server_info = sqlsrv_server_info( $this->mConn );
 627                 $version = 'Error';
 628                 if ( isset( $server_info['SQLServerVersion'] ) ) {
 629                         $version = $server_info['SQLServerVersion'];
 630                 }
 631                 return $version;
 632         }
 633
 634         function tableExists ( $table, $fname = __METHOD__, $schema = false ) {
 635                 $res = sqlsrv_query( $this->mConn, "SELECT * FROM information_schema.tables
 636                         WHERE table_type='BASE TABLE' AND table_name = '$table'" );
 637                 if ( $res === false ) {
 638                         print( "Error in tableExists query: " . $this->getErrors() );
 639                         return false;
 640                 }
 641                 if ( sqlsrv_fetch( $res ) ) {
 642                         return true;
 643                 } else {
 644                         return false;
 645                 }
 646         }
 647
 648         /**
 649          * Query whether a given column exists in the mediawiki schema
 650          */
 651         function fieldExists( $table, $field, $fname = 'DatabaseMssql::fieldExists' ) {
 652                 $table = $this->tableName( $table );
 653                 $res = sqlsrv_query( $this->mConn, "SELECT DATA_TYPE FROM INFORMATION_SCHEMA.Columns
 654                         WHERE TABLE_NAME = '$table' AND COLUMN_NAME = '$field'" );
 655                 if ( $res === false ) {
 656                         print( "Error in fieldExists query: " . $this->getErrors() );
 657                         return false;
 658                 }
 659                 if ( sqlsrv_fetch( $res ) ) {
 660                         return true;
 661                 } else {
 662                         return false;
 663                 }
 664         }
 665
 666         function fieldInfo( $table, $field ) {
 667                 $table = $this->tableName( $table );
 668                 $res = sqlsrv_query( $this->mConn, "SELECT * FROM INFORMATION_SCHEMA.Columns
 669                         WHERE TABLE_NAME = '$table' AND COLUMN_NAME = '$field'" );
 670                 if ( $res === false ) {
 671                         print( "Error in fieldInfo query: " . $this->getErrors() );
 672                         return false;
 673                 }
 674                 $meta = $this->fetchRow( $res );
 675                 if ( $meta ) {
 676                         return new MssqlField( $meta );
 677                 }
 678                 return false;
 679         }
 680
 681         /**
 682          * Begin a transaction, committing any previously open transaction
 683          */
 684         function begin( $fname = 'DatabaseMssql::begin' ) {
 685                 sqlsrv_begin_transaction( $this->mConn );
 686                 $this->mTrxLevel = 1;
 687         }
 688
 689         /**
 690          * End a transaction
 691          */
 692         function commit( $fname = 'DatabaseMssql::commit' ) {
 693                 sqlsrv_commit( $this->mConn );
 694                 $this->mTrxLevel = 0;
 695         }
 696
 697         /**
 698          * Rollback a transaction.
 699          * No-op on non-transactional databases.
 700          */
 701         function rollback( $fname = 'DatabaseMssql::rollback' ) {
 702                 sqlsrv_rollback( $this->mConn );
 703                 $this->mTrxLevel = 0;
 704         }
 705
 706         /**
 707          * Escapes a identifier for use inm SQL.
 708          * Throws an exception if it is invalid.
 709          * Reference: http://msdn.microsoft.com/en-us/library/aa224033%28v=SQL.80%29.aspx
 710          */
 711         private function escapeIdentifier( $identifier ) {
 712                 if ( strlen( $identifier ) == 0 ) {
 713                         throw new MWException( "An identifier must not be empty" );
 714                 }
 715                 if ( strlen( $identifier ) > 128 ) {
 716                         throw new MWException( "The identifier '$identifier' is too long (max. 128)" );
 717                 }
 718                 if ( ( strpos( $identifier, '[' ) !== false ) || ( strpos( $identifier, ']' ) !== false ) ) {
 719                         // It may be allowed if you quoted with double quotation marks, but that would break if QUOTED_IDENTIFIER is OFF
 720                         throw new MWException( "You can't use square brackers in the identifier '$identifier'" );
 721                 }
 722                 return "[$identifier]";
 723         }
 724
 725         /**
 726          * Initial setup.
 727          * Precondition: This object is connected as the superuser.
 728          * Creates the database, schema, user and login.
 729          */
 730         function initial_setup( $dbName, $newUser, $loginPassword ) {
 731                 $dbName = $this->escapeIdentifier( $dbName );
 732
 733                 // It is not clear what can be used as a login,
 734                 // From http://msdn.microsoft.com/en-us/library/ms173463.aspx
 735                 // a sysname may be the same as an identifier.
 736                 $newUser = $this->escapeIdentifier( $newUser );
 737                 $loginPassword = $this->addQuotes( $loginPassword );
 738
 739                 $this->doQuery("CREATE DATABASE $dbName;");
 740                 $this->doQuery("USE $dbName;");
 741                 $this->doQuery("CREATE SCHEMA $dbName;");
 742                 $this->doQuery("
 743                                                 CREATE
 744                                                         LOGIN $newUser
 745                                                 WITH
 746                                                         PASSWORD=$loginPassword
 747                                                 ;
 748                                         ");
 749                 $this->doQuery("
 750                                                 CREATE
 751                                                         USER $newUser
 752                                                 FOR
 753                                                         LOGIN $newUser
 754                                                 WITH
 755                                                         DEFAULT_SCHEMA=$dbName
 756                                                 ;
 757                                         ");
 758                 $this->doQuery("
 759                                                 GRANT
 760                                                         BACKUP DATABASE,
 761                                                         BACKUP LOG,
 762                                                         CREATE DEFAULT,
 763                                                         CREATE FUNCTION,
 764                                                         CREATE PROCEDURE,
 765                                                         CREATE RULE,
 766                                                         CREATE TABLE,
 767                                                         CREATE VIEW,
 768                                                         CREATE FULLTEXT CATALOG
 769                                                 ON
 770                                                         DATABASE::$dbName
 771                                                 TO $newUser
 772                                                 ;
 773                                         ");
 774                 $this->doQuery("
 775                                                 GRANT
 776                                                         CONTROL
 777                                                 ON
 778                                                         SCHEMA::$dbName
 779                                                 TO $newUser
 780                                                 ;
 781                                         ");
 782
 783
 784         }
 785
 786         function encodeBlob( $b ) {
 787         // we can't have zero's and such, this is a simple encoding to make sure we don't barf
 788                 return base64_encode( $b );
 789         }
 790
 791         function decodeBlob( $b ) {
 792         // we can't have zero's and such, this is a simple encoding to make sure we don't barf
 793         return base64_decode( $b );
 794         }
 795
 796         /**
 797          * @private
 798          */
 799         function tableNamesWithUseIndexOrJOIN( $tables, $use_index = array(), $join_conds = array() ) {
 800                 $ret = array();
 801                 $retJOIN = array();
 802                 $use_index_safe = is_array( $use_index ) ? $use_index : array();
 803                 $join_conds_safe = is_array( $join_conds ) ? $join_conds : array();
 804                 foreach ( $tables as $table ) {
 805                         // Is there a JOIN and INDEX clause for this table?
 806                         if ( isset( $join_conds_safe[$table] ) && isset( $use_index_safe[$table] ) ) {
 807                                 $tableClause = $join_conds_safe[$table][0] . ' ' . $this->tableName( $table );
 808                                 $tableClause .= ' ' . $this->useIndexClause( implode( ',', (array)$use_index_safe[$table] ) );
 809                                 $tableClause .= ' ON (' . $this->makeList( (array)$join_conds_safe[$table][1], LIST_AND ) . ')';
 810                                 $retJOIN[] = $tableClause;
 811                         // Is there an INDEX clause?
 812                         } elseif ( isset( $use_index_safe[$table] ) ) {
 813                                 $tableClause = $this->tableName( $table );
 814                                 $tableClause .= ' ' . $this->useIndexClause( implode( ',', (array)$use_index_safe[$table] ) );
 815                                 $ret[] = $tableClause;
 816                         // Is there a JOIN clause?
 817                         } elseif ( isset( $join_conds_safe[$table] ) ) {
 818                                 $tableClause = $join_conds_safe[$table][0] . ' ' . $this->tableName( $table );
 819                                 $tableClause .= ' ON (' . $this->makeList( (array)$join_conds_safe[$table][1], LIST_AND ) . ')';
 820                                 $retJOIN[] = $tableClause;
 821                         } else {
 822                                 $tableClause = $this->tableName( $table );
 823                                 $ret[] = $tableClause;
 824                         }
 825                 }
 826                 // We can't separate explicit JOIN clauses with ',', use ' ' for those
 827                 $straightJoins = !empty( $ret ) ? implode( ',', $ret ) : "";
 828                 $otherJoins = !empty( $retJOIN ) ? implode( ' ', $retJOIN ) : "";
 829                 // Compile our final table clause
 830                 return implode( ' ', array( $straightJoins, $otherJoins ) );
 831         }
 832
 833         function strencode( $s ) { # Should not be called by us
 834                 return str_replace( "'", "''", $s );
 835         }
 836
 837         function addQuotes( $s ) {
 838                 if ( $s instanceof Blob ) {
 839                         return "'" . $s->fetch( $s ) . "'";
 840                 } else {
 841                         return parent::addQuotes( $s );
 842                 }
 843         }
 844
 845         public function addIdentifierQuotes( $s ) {
 846                 // http://msdn.microsoft.com/en-us/library/aa223962.aspx
 847                 return '[' . $s . ']';
 848         }
 849
 850         public function isQuotedIdentifier( $name ) {
 851                 return $name[0] == '[' && substr( $name, -1, 1 ) == ']';
 852         }
 853
 854         function selectDB( $db ) {
 855                 return ( $this->query( "SET DATABASE $db" ) !== false );
 856         }
 857
 858         /**
 859          * @private
 860          *
 861          * @param $options Array: an associative array of options to be turned into
 862          *                 an SQL query, valid keys are listed in the function.
 863          * @return Array
 864          */
 865         function makeSelectOptions( $options ) {
 866                 $tailOpts = '';
 867                 $startOpts = '';
 868
 869                 $noKeyOptions = array();
 870                 foreach ( $options as $key => $option ) {
 871                         if ( is_numeric( $key ) ) {
 872                                 $noKeyOptions[$option] = true;
 873                         }
 874                 }
 875
 876                 if ( isset( $options['GROUP BY'] ) ) {
 877                         $tailOpts .= " GROUP BY {$options['GROUP BY']}";
 878                 }
 879                 if ( isset( $options['HAVING'] ) ) {
 880                         $tailOpts .= " HAVING {$options['GROUP BY']}";
 881                 }
 882                 if ( isset( $options['ORDER BY'] ) ) {
 883                         $tailOpts .= " ORDER BY {$options['ORDER BY']}";
 884                 }
 885
 886                 if ( isset( $noKeyOptions['DISTINCT'] ) && isset( $noKeyOptions['DISTINCTROW'] ) ) {
 887                         $startOpts .= 'DISTINCT';
 888                 }
 889
 890                 // we want this to be compatible with the output of parent::makeSelectOptions()
 891                 return array( $startOpts, '' , $tailOpts, '' );
 892         }
 893
 894         /**
 895          * Get the type of the DBMS, as it appears in $wgDBtype.
 896          */
 897         function getType(){
 898                 return 'mssql';
 899         }
 900
 901         function buildConcat( $stringList ) {
 902                 return implode( ' + ', $stringList );
 903         }
 904
 905         public function getSearchEngine() {
 906                 return "SearchMssql";
 907         }
 908
 909         /**
 910          * Since MSSQL doesn't recognize the infinity keyword, set date manually.
 911          * @todo Remove magic date
 912          */
 913         public function getInfinity() {
 914                 return '3000-01-31 00:00:00.000';
 915         }
 916
 917 } // end DatabaseMssql class
 918
 919 /**
 920  * Utility class.
 921  *
 922  * @ingroup Database
 923  */
 924 class MssqlField implements Field {
 925         private $name, $tablename, $default, $max_length, $nullable, $type;
 926         function __construct ( $info ) {
 927                 $this->name = $info['COLUMN_NAME'];
 928                 $this->tablename = $info['TABLE_NAME'];
 929                 $this->default = $info['COLUMN_DEFAULT'];
 930                 $this->max_length = $info['CHARACTER_MAXIMUM_LENGTH'];
 931                 $this->nullable = !( strtolower( $info['IS_NULLABLE'] ) == 'no' );
 932                 $this->type = $info['DATA_TYPE'];
 933         }
 934
 935         function name() {
 936                 return $this->name;
 937         }
 938
 939         function tableName() {
 940                 return $this->tableName;
 941         }
 942
 943         function defaultValue() {
 944                 return $this->default;
 945         }
 946
 947         function maxLength() {
 948                 return $this->max_length;
 949         }
 950
 951         function isNullable() {
 952                 return $this->nullable;
 953         }
 954
 955         function type() {
 956                 return $this->type;
 957         }
 958 }
 959
 960 /**
 961  * The MSSQL PHP driver doesn't support sqlsrv_num_rows, so we recall all rows into an array and maintain our
 962  * own cursor index into that array...This is similar to the way the Oracle driver handles this same issue
 963  *
 964  * @ingroup Database
 965  */
 966 class MssqlResult {
 967
 968         public function __construct( $queryresult = false ) {
 969                 $this->mCursor = 0;
 970                 $this->mRows = array();
 971                 $this->mNumFields = sqlsrv_num_fields( $queryresult );
 972                 $this->mFieldMeta = sqlsrv_field_metadata( $queryresult );
 973
 974                 $rows = sqlsrv_fetch_array( $queryresult, SQLSRV_FETCH_ASSOC );
 975
 976                 foreach( $rows as $row ) {
 977                         if ( $row !== null ) {
 978                                 foreach ( $row as $k => $v ) {
 979                                         if ( is_object( $v ) && method_exists( $v, 'format' ) ) {// DateTime Object
 980                                                 $row[$k] = $v->format( "Y-m-d\TH:i:s\Z" );
 981                                         }
 982                                 }
 983                                 $this->mRows[] = $row;// read results into memory, cursors are not supported
 984                         }
 985                 }
 986                 $this->mRowCount = count( $this->mRows );
 987                 sqlsrv_free_stmt( $queryresult );
 988         }
 989
 990         private function array_to_obj( $array, &$obj ) {
 991                 foreach ( $array as $key => $value ) {
 992                         if ( is_array( $value ) ) {
 993                                 $obj->$key = new stdClass();
 994                                 $this->array_to_obj( $value, $obj->$key );
 995                         } else {
 996                                 if ( !empty( $key ) ) {
 997                                         $obj->$key = $value;
 998                                 }
 999                         }
1000                 }
1001                 return $obj;
1002         }
1003
1004         public function fetch( $mode = SQLSRV_FETCH_BOTH, $object_class = 'stdClass' ) {
1005                 if ( $this->mCursor >= $this->mRowCount || $this->mRowCount == 0 ) {
1006                         return false;
1007                 }
1008                 $arrNum = array();
1009                 if ( $mode == SQLSRV_FETCH_NUMERIC || $mode == SQLSRV_FETCH_BOTH ) {
1010                         foreach ( $this->mRows[$this->mCursor] as $value ) {
1011                                 $arrNum[] = $value;
1012                         }
1013                 }
1014                 switch( $mode ) {
1015                         case SQLSRV_FETCH_ASSOC:
1016                                 $ret = $this->mRows[$this->mCursor];
1017                                 break;
1018                         case SQLSRV_FETCH_NUMERIC:
1019                                 $ret = $arrNum;
1020                                 break;
1021                         case 'OBJECT':
1022                                 $o = new $object_class;
1023                                 $ret = $this->array_to_obj( $this->mRows[$this->mCursor], $o );
1024                                 break;
1025                         case SQLSRV_FETCH_BOTH:
1026                         default:
1027                                 $ret = $this->mRows[$this->mCursor] + $arrNum;
1028                                 break;
1029                 }
1030
1031                 $this->mCursor++;
1032                 return $ret;
1033         }
1034
1035         public function get( $pos, $fld ) {
1036                 return $this->mRows[$pos][$fld];
1037         }
1038
1039         public function numrows() {
1040                 return $this->mRowCount;
1041         }
1042
1043         public function seek( $iRow ) {
1044                 $this->mCursor = min( $iRow, $this->mRowCount );
1045         }
1046
1047         public function numfields() {
1048                 return $this->mNumFields;
1049         }
1050
1051         public function fieldname( $nr ) {
1052                 $arrKeys = array_keys( $this->mRows[0] );
1053                 return $arrKeys[$nr];
1054         }
1055
1056         public function fieldtype( $nr ) {
1057                 $i = 0;
1058                 $intType = -1;
1059                 foreach ( $this->mFieldMeta as $meta ) {
1060                         if ( $nr == $i ) {
1061                                 $intType = $meta['Type'];
1062                                 break;
1063                         }
1064                         $i++;
1065                 }
1066                 // http://msdn.microsoft.com/en-us/library/cc296183.aspx contains type table
1067                 switch( $intType ) {
1068                         case SQLSRV_SQLTYPE_BIGINT:             $strType = 'bigint'; break;
1069                         case SQLSRV_SQLTYPE_BINARY:             $strType = 'binary'; break;
1070                         case SQLSRV_SQLTYPE_BIT:                        $strType = 'bit'; break;
1071                         case SQLSRV_SQLTYPE_CHAR:                       $strType = 'char'; break;
1072                         case SQLSRV_SQLTYPE_DATETIME:           $strType = 'datetime'; break;
1073                         case SQLSRV_SQLTYPE_DECIMAL/*($precision, $scale)*/: $strType = 'decimal'; break;
1074                         case SQLSRV_SQLTYPE_FLOAT:                      $strType = 'float'; break;
1075                         case SQLSRV_SQLTYPE_IMAGE:                      $strType = 'image'; break;
1076                         case SQLSRV_SQLTYPE_INT:                        $strType = 'int'; break;
1077                         case SQLSRV_SQLTYPE_MONEY:                      $strType = 'money'; break;
1078                         case SQLSRV_SQLTYPE_NCHAR/*($charCount)*/: $strType = 'nchar'; break;
1079                         case SQLSRV_SQLTYPE_NUMERIC/*($precision, $scale)*/: $strType = 'numeric'; break;
1080                         case SQLSRV_SQLTYPE_NVARCHAR/*($charCount)*/: $strType = 'nvarchar'; break;
1081                         // case SQLSRV_SQLTYPE_NVARCHAR('max'): $strType = 'nvarchar(MAX)'; break;
1082                         case SQLSRV_SQLTYPE_NTEXT:                      $strType = 'ntext'; break;
1083                         case SQLSRV_SQLTYPE_REAL:                       $strType = 'real'; break;
1084                         case SQLSRV_SQLTYPE_SMALLDATETIME:      $strType = 'smalldatetime'; break;
1085                         case SQLSRV_SQLTYPE_SMALLINT:           $strType = 'smallint'; break;
1086                         case SQLSRV_SQLTYPE_SMALLMONEY:         $strType = 'smallmoney'; break;
1087                         case SQLSRV_SQLTYPE_TEXT:                       $strType = 'text'; break;
1088                         case SQLSRV_SQLTYPE_TIMESTAMP:          $strType = 'timestamp'; break;
1089                         case SQLSRV_SQLTYPE_TINYINT:            $strType = 'tinyint'; break;
1090                         case SQLSRV_SQLTYPE_UNIQUEIDENTIFIER: $strType = 'uniqueidentifier'; break;
1091                         case SQLSRV_SQLTYPE_UDT:                        $strType = 'UDT'; break;
1092                         case SQLSRV_SQLTYPE_VARBINARY/*($byteCount)*/: $strType = 'varbinary'; break;
1093                         // case SQLSRV_SQLTYPE_VARBINARY('max'): $strType = 'varbinary(MAX)'; break;
1094                         case SQLSRV_SQLTYPE_VARCHAR/*($charCount)*/: $strType = 'varchar'; break;
1095                         // case SQLSRV_SQLTYPE_VARCHAR('max'): $strType = 'varchar(MAX)'; break;
1096                         case SQLSRV_SQLTYPE_XML:                        $strType = 'xml'; break;
1097                         default: $strType = $intType;
1098                 }
1099                 return $strType;
1100         }
1101
1102         public function free() {
1103                 unset( $this->mRows );
1104                 return;
1105         }
1106 }