DatabaseMssql.php

Go to the documentation of this file.

<?php
class DatabaseMssql extends DatabaseBase {
    protected $mInsertId = null;
    protected $mLastResult = null;
    protected $mAffectedRows = null;
    protected $mSubqueryId = 0;
    protected $mScrollableCursor = true;
    protected $mPrepareStatements = true;
    protected $mBinaryColumnCache = null;
    protected $mBitColumnCache = null;
    protected $mIgnoreDupKeyErrors = false;
    protected $mIgnoreErrors = [];

    protected $mPort;

    public function implicitGroupby() {
        return false;
    }

    public function implicitOrderby() {
        return false;
    }

    public function unionSupportsOrderAndLimit() {
        return false;
    }

    public function open( $server, $user, $password, $dbName ) {
        # Test for driver support, to avoid suppressed fatal error
        if ( !function_exists( 'sqlsrv_connect' ) ) {
            throw new DBConnectionError(
                $this,
                "Microsoft SQL Server Native (sqlsrv) functions missing.
                You can download the driver from: http://go.microsoft.com/fwlink/?LinkId=123470\n"
            );
        }

        global $wgDBport, $wgDBWindowsAuthentication;

        # e.g. the class is being loaded
        if ( !strlen( $user ) ) {
            return null;
        }

        $this->close();
        $this->mServer = $server;
        $this->mPort = $wgDBport;
        $this->mUser = $user;
        $this->mPassword = $password;
        $this->mDBname = $dbName;

        $connectionInfo = [];

        if ( $dbName ) {
            $connectionInfo['Database'] = $dbName;
        }

        // Decide which auth scenerio to use
        // if we are using Windows auth, don't add credentials to $connectionInfo
        if ( !$wgDBWindowsAuthentication ) {
            $connectionInfo['UID'] = $user;
            $connectionInfo['PWD'] = $password;
        }

        MediaWiki\suppressWarnings();
        $this->mConn = sqlsrv_connect( $server, $connectionInfo );
        MediaWiki\restoreWarnings();

        if ( $this->mConn === false ) {
            throw new DBConnectionError( $this, $this->lastError() );
        }

        $this->mOpened = true;

        return $this->mConn;
    }

    protected function closeConnection() {
        return sqlsrv_close( $this->mConn );
    }

    protected function resultObject( $result ) {
        if ( !$result ) {
            return false;
        } elseif ( $result instanceof MssqlResultWrapper ) {
            return $result;
        } elseif ( $result === true ) {
            // Successful write query
            return $result;
        } else {
            return new MssqlResultWrapper( $this, $result );
        }
    }

    protected function doQuery( $sql ) {
        if ( $this->getFlag( DBO_DEBUG ) ) {
            wfDebug( "SQL: [$sql]\n" );
        }
        $this->offset = 0;

        // several extensions seem to think that all databases support limits
        // via LIMIT N after the WHERE clause well, MSSQL uses SELECT TOP N,
        // so to catch any of those extensions we'll do a quick check for a
        // LIMIT clause and pass $sql through $this->LimitToTopN() which parses
        // the limit clause and passes the result to $this->limitResult();
        if ( preg_match( '/\bLIMIT\s*/i', $sql ) ) {
            // massage LIMIT -> TopN
            $sql = $this->LimitToTopN( $sql );
        }

        // MSSQL doesn't have EXTRACT(epoch FROM XXX)
        if ( preg_match( '#\bEXTRACT\s*?\(\s*?EPOCH\s+FROM\b#i', $sql, $matches ) ) {
            // This is same as UNIX_TIMESTAMP, we need to calc # of seconds from 1970
            $sql = str_replace( $matches[0], "DATEDIFF(s,CONVERT(datetime,'1/1/1970'),", $sql );
        }

        // perform query

        // SQLSRV_CURSOR_STATIC is slower than SQLSRV_CURSOR_CLIENT_BUFFERED (one of the two is
        // needed if we want to be able to seek around the result set), however CLIENT_BUFFERED
        // has a bug in the sqlsrv driver where wchar_t types (such as nvarchar) that are empty
        // strings make php throw a fatal error "Severe error translating Unicode"
        if ( $this->mScrollableCursor ) {
            $scrollArr = [ 'Scrollable' => SQLSRV_CURSOR_STATIC ];
        } else {
            $scrollArr = [];
        }

        if ( $this->mPrepareStatements ) {
            // we do prepare + execute so we can get its field metadata for later usage if desired
            $stmt = sqlsrv_prepare( $this->mConn, $sql, [], $scrollArr );
            $success = sqlsrv_execute( $stmt );
        } else {
            $stmt = sqlsrv_query( $this->mConn, $sql, [], $scrollArr );
            $success = (bool)$stmt;
        }

        // make a copy so that anything we add below does not get reflected in future queries
        $ignoreErrors = $this->mIgnoreErrors;

        if ( $this->mIgnoreDupKeyErrors ) {
            // ignore duplicate key errors
            // this emulates INSERT IGNORE in MySQL
            $ignoreErrors[] = '2601'; // duplicate key error caused by unique index
            $ignoreErrors[] = '2627'; // duplicate key error caused by primary key
            $ignoreErrors[] = '3621'; // generic "the statement has been terminated" error
        }

        if ( $success === false ) {
            $errors = sqlsrv_errors();
            $success = true;

            foreach ( $errors as $err ) {
                if ( !in_array( $err['code'], $ignoreErrors ) ) {
                    $success = false;
                    break;
                }
            }

            if ( $success === false ) {
                return false;
            }
        }
        // remember number of rows affected
        $this->mAffectedRows = sqlsrv_rows_affected( $stmt );

        return $stmt;
    }

    public function freeResult( $res ) {
        if ( $res instanceof ResultWrapper ) {
            $res = $res->result;
        }

        sqlsrv_free_stmt( $res );
    }

    public function fetchObject( $res ) {
        // $res is expected to be an instance of MssqlResultWrapper here
        return $res->fetchObject();
    }

    public function fetchRow( $res ) {
        return $res->fetchRow();
    }

    public function numRows( $res ) {
        if ( $res instanceof ResultWrapper ) {
            $res = $res->result;
        }

        $ret = sqlsrv_num_rows( $res );

        if ( $ret === false ) {
            // we cannot get an amount of rows from this cursor type
            // has_rows returns bool true/false if the result has rows
            $ret = (int)sqlsrv_has_rows( $res );
        }

        return $ret;
    }

    public function numFields( $res ) {
        if ( $res instanceof ResultWrapper ) {
            $res = $res->result;
        }

        return sqlsrv_num_fields( $res );
    }

    public function fieldName( $res, $n ) {
        if ( $res instanceof ResultWrapper ) {
            $res = $res->result;
        }

        return sqlsrv_field_metadata( $res )[$n]['Name'];
    }

    public function insertId() {
        return $this->mInsertId;
    }

    public function dataSeek( $res, $row ) {
        return $res->seek( $row );
    }

    public function lastError() {
        $strRet = '';
        $retErrors = sqlsrv_errors( SQLSRV_ERR_ALL );
        if ( $retErrors != null ) {
            foreach ( $retErrors as $arrError ) {
                $strRet .= $this->formatError( $arrError ) . "\n";
            }
        } else {
            $strRet = "No errors found";
        }

        return $strRet;
    }

    private function formatError( $err ) {
        return '[SQLSTATE ' . $err['SQLSTATE'] . '][Error Code ' . $err['code'] . ']' . $err['message'];
    }

    public function lastErrno() {
        $err = sqlsrv_errors( SQLSRV_ERR_ALL );
        if ( $err !== null && isset( $err[0] ) ) {
            return $err[0]['code'];
        } else {
            return 0;
        }
    }

    public function affectedRows() {
        return $this->mAffectedRows;
    }

    public function select( $table, $vars, $conds = '', $fname = __METHOD__,
        $options = [], $join_conds = []
    ) {
        $sql = $this->selectSQLText( $table, $vars, $conds, $fname, $options, $join_conds );
        if ( isset( $options['EXPLAIN'] ) ) {
            try {
                $this->mScrollableCursor = false;
                $this->mPrepareStatements = false;
                $this->query( "SET SHOWPLAN_ALL ON" );
                $ret = $this->query( $sql, $fname );
                $this->query( "SET SHOWPLAN_ALL OFF" );
            } catch ( DBQueryError $dqe ) {
                if ( isset( $options['FOR COUNT'] ) ) {
                    // likely don't have privs for SHOWPLAN, so run a select count instead
                    $this->query( "SET SHOWPLAN_ALL OFF" );
                    unset( $options['EXPLAIN'] );
                    $ret = $this->select(
                        $table,
                        'COUNT(*) AS EstimateRows',
                        $conds,
                        $fname,
                        $options,
                        $join_conds
                    );
                } else {
                    // someone actually wanted the query plan instead of an est row count
                    // let them know of the error
                    $this->mScrollableCursor = true;
                    $this->mPrepareStatements = true;
                    throw $dqe;
                }
            }
            $this->mScrollableCursor = true;
            $this->mPrepareStatements = true;
            return $ret;
        }
        return $this->query( $sql, $fname );
    }

    public function selectSQLText( $table, $vars, $conds = '', $fname = __METHOD__,
        $options = [], $join_conds = []
    ) {
        if ( isset( $options['EXPLAIN'] ) ) {
            unset( $options['EXPLAIN'] );
        }

        $sql = parent::selectSQLText( $table, $vars, $conds, $fname, $options, $join_conds );

        // try to rewrite aggregations of bit columns (currently MAX and MIN)
        if ( strpos( $sql, 'MAX(' ) !== false || strpos( $sql, 'MIN(' ) !== false ) {
            $bitColumns = [];
            if ( is_array( $table ) ) {
                foreach ( $table as $t ) {
                    $bitColumns += $this->getBitColumns( $this->tableName( $t ) );
                }
            } else {
                $bitColumns = $this->getBitColumns( $this->tableName( $table ) );
            }

            foreach ( $bitColumns as $col => $info ) {
                $replace = [
                    "MAX({$col})" => "MAX(CAST({$col} AS tinyint))",
                    "MIN({$col})" => "MIN(CAST({$col} AS tinyint))",
                ];
                $sql = str_replace( array_keys( $replace ), array_values( $replace ), $sql );
            }
        }

        return $sql;
    }

    public function deleteJoin( $delTable, $joinTable, $delVar, $joinVar, $conds,
        $fname = __METHOD__
    ) {
        $this->mScrollableCursor = false;
        try {
            parent::deleteJoin( $delTable, $joinTable, $delVar, $joinVar, $conds, $fname );
        } catch ( Exception $e ) {
            $this->mScrollableCursor = true;
            throw $e;
        }
        $this->mScrollableCursor = true;
    }

    public function delete( $table, $conds, $fname = __METHOD__ ) {
        $this->mScrollableCursor = false;
        try {
            parent::delete( $table, $conds, $fname );
        } catch ( Exception $e ) {
            $this->mScrollableCursor = true;
            throw $e;
        }
        $this->mScrollableCursor = true;
    }

    public function estimateRowCount( $table, $vars = '*', $conds = '',
        $fname = __METHOD__, $options = []
    ) {
        // http://msdn2.microsoft.com/en-us/library/aa259203.aspx
        $options['EXPLAIN'] = true;
        $options['FOR COUNT'] = true;
        $res = $this->select( $table, $vars, $conds, $fname, $options );

        $rows = -1;
        if ( $res ) {
            $row = $this->fetchRow( $res );

            if ( isset( $row['EstimateRows'] ) ) {
                $rows = (int)$row['EstimateRows'];
            }
        }

        return $rows;
    }

    public function indexInfo( $table, $index, $fname = __METHOD__ ) {
        # This does not return the same info as MYSQL would, but that's OK
        # because MediaWiki never uses the returned value except to check for
        # the existance of indexes.
        $sql = "sp_helpindex '" . $this->tableName( $table ) . "'";
        $res = $this->query( $sql, $fname );

        if ( !$res ) {
            return null;
        }

        $result = [];
        foreach ( $res as $row ) {
            if ( $row->index_name == $index ) {
                $row->Non_unique = !stristr( $row->index_description, "unique" );
                $cols = explode( ", ", $row->index_keys );
                foreach ( $cols as $col ) {
                    $row->Column_name = trim( $col );
                    $result[] = clone $row;
                }
            } elseif ( $index == 'PRIMARY' && stristr( $row->index_description, 'PRIMARY' ) ) {
                $row->Non_unique = 0;
                $cols = explode( ", ", $row->index_keys );
                foreach ( $cols as $col ) {
                    $row->Column_name = trim( $col );
                    $result[] = clone $row;
                }
            }
        }

        return empty( $result ) ? false : $result;
    }

    public function insert( $table, $arrToInsert, $fname = __METHOD__, $options = [] ) {
        # No rows to insert, easy just return now
        if ( !count( $arrToInsert ) ) {
            return true;
        }

        if ( !is_array( $options ) ) {
            $options = [ $options ];
        }

        $table = $this->tableName( $table );

        if ( !( isset( $arrToInsert[0] ) && is_array( $arrToInsert[0] ) ) ) { // Not multi row
            $arrToInsert = [ 0 => $arrToInsert ]; // make everything multi row compatible
        }

        // We know the table we're inserting into, get its identity column
        $identity = null;
        // strip matching square brackets and the db/schema from table name
        $tableRawArr = explode( '.', preg_replace( '#\[([^\]]*)\]#', '$1', $table ) );
        $tableRaw = array_pop( $tableRawArr );
        $res = $this->doQuery(
            "SELECT NAME AS idColumn FROM SYS.IDENTITY_COLUMNS " .
                "WHERE OBJECT_NAME(OBJECT_ID)='{$tableRaw}'"
        );
        if ( $res && sqlsrv_has_rows( $res ) ) {
            // There is an identity for this table.
            $identityArr = sqlsrv_fetch_array( $res, SQLSRV_FETCH_ASSOC );
            $identity = array_pop( $identityArr );
        }
        sqlsrv_free_stmt( $res );

        // Determine binary/varbinary fields so we can encode data as a hex string like 0xABCDEF
        $binaryColumns = $this->getBinaryColumns( $table );

        // INSERT IGNORE is not supported by SQL Server
        // remove IGNORE from options list and set ignore flag to true
        if ( in_array( 'IGNORE', $options ) ) {
            $options = array_diff( $options, [ 'IGNORE' ] );
            $this->mIgnoreDupKeyErrors = true;
        }

        foreach ( $arrToInsert as $a ) {
            // start out with empty identity column, this is so we can return
            // it as a result of the insert logic
            $sqlPre = '';
            $sqlPost = '';
            $identityClause = '';

            // if we have an identity column
            if ( $identity ) {
                // iterate through
                foreach ( $a as $k => $v ) {
                    if ( $k == $identity ) {
                        if ( !is_null( $v ) ) {
                            // there is a value being passed to us,
                            // we need to turn on and off inserted identity
                            $sqlPre = "SET IDENTITY_INSERT $table ON;";
                            $sqlPost = ";SET IDENTITY_INSERT $table OFF;";
                        } else {
                            // we can't insert NULL into an identity column,
                            // so remove the column from the insert.
                            unset( $a[$k] );
                        }
                    }
                }

                // we want to output an identity column as result
                $identityClause = "OUTPUT INSERTED.$identity ";
            }

            $keys = array_keys( $a );

            // Build the actual query
            $sql = $sqlPre . 'INSERT ' . implode( ' ', $options ) .
                " INTO $table (" . implode( ',', $keys ) . ") $identityClause VALUES (";

            $first = true;
            foreach ( $a as $key => $value ) {
                if ( isset( $binaryColumns[$key] ) ) {
                    $value = new MssqlBlob( $value );
                }
                if ( $first ) {
                    $first = false;
                } else {
                    $sql .= ',';
                }
                if ( is_null( $value ) ) {
                    $sql .= 'null';
                } elseif ( is_array( $value ) || is_object( $value ) ) {
                    if ( is_object( $value ) && $value instanceof Blob ) {
                        $sql .= $this->addQuotes( $value );
                    } else {
                        $sql .= $this->addQuotes( serialize( $value ) );
                    }
                } else {
                    $sql .= $this->addQuotes( $value );
                }
            }
            $sql .= ')' . $sqlPost;

            // Run the query
            $this->mScrollableCursor = false;
            try {
                $ret = $this->query( $sql );
            } catch ( Exception $e ) {
                $this->mScrollableCursor = true;
                $this->mIgnoreDupKeyErrors = false;
                throw $e;
            }
            $this->mScrollableCursor = true;

            if ( !is_null( $identity ) ) {
                // then we want to get the identity column value we were assigned and save it off
                $row = $ret->fetchObject();
                if ( is_object( $row ) ) {
                    $this->mInsertId = $row->$identity;

                    // it seems that mAffectedRows is -1 sometimes when OUTPUT INSERTED.identity is used
                    // if we got an identity back, we know for sure a row was affected, so adjust that here
                    if ( $this->mAffectedRows == -1 ) {
                        $this->mAffectedRows = 1;
                    }
                }
            }
        }
        $this->mIgnoreDupKeyErrors = false;
        return $ret;
    }

    public function nativeInsertSelect( $destTable, $srcTable, $varMap, $conds, $fname = __METHOD__,
        $insertOptions = [], $selectOptions = []
    ) {
        $this->mScrollableCursor = false;
        try {
            $ret = parent::nativeInsertSelect(
                $destTable,
                $srcTable,
                $varMap,
                $conds,
                $fname,
                $insertOptions,
                $selectOptions
            );
        } catch ( Exception $e ) {
            $this->mScrollableCursor = true;
            throw $e;
        }
        $this->mScrollableCursor = true;

        return $ret;
    }

    function update( $table, $values, $conds, $fname = __METHOD__, $options = [] ) {
        $table = $this->tableName( $table );
        $binaryColumns = $this->getBinaryColumns( $table );

        $opts = $this->makeUpdateOptions( $options );
        $sql = "UPDATE $opts $table SET " . $this->makeList( $values, LIST_SET, $binaryColumns );

        if ( $conds !== [] && $conds !== '*' ) {
            $sql .= " WHERE " . $this->makeList( $conds, LIST_AND, $binaryColumns );
        }

        $this->mScrollableCursor = false;
        try {
            $this->query( $sql );
        } catch ( Exception $e ) {
            $this->mScrollableCursor = true;
            throw $e;
        }
        $this->mScrollableCursor = true;
        return true;
    }

    public function makeList( $a, $mode = LIST_COMMA, $binaryColumns = [] ) {
        if ( !is_array( $a ) ) {
            throw new DBUnexpectedError( $this, __METHOD__ . ' called with incorrect parameters' );
        }

        if ( $mode != LIST_NAMES ) {
            // In MS SQL, values need to be specially encoded when they are
            // inserted into binary fields. Perform this necessary encoding
            // for the specified set of columns.
            foreach ( array_keys( $a ) as $field ) {
                if ( !isset( $binaryColumns[$field] ) ) {
                    continue;
                }

                if ( is_array( $a[$field] ) ) {
                    foreach ( $a[$field] as &$v ) {
                        $v = new MssqlBlob( $v );
                    }
                    unset( $v );
                } else {
                    $a[$field] = new MssqlBlob( $a[$field] );
                }
            }
        }

        return parent::makeList( $a, $mode );
    }

    public function textFieldSize( $table, $field ) {
        $table = $this->tableName( $table );
        $sql = "SELECT CHARACTER_MAXIMUM_LENGTH,DATA_TYPE FROM INFORMATION_SCHEMA.Columns
            WHERE TABLE_NAME = '$table' AND COLUMN_NAME = '$field'";
        $res = $this->query( $sql );
        $row = $this->fetchRow( $res );
        $size = -1;
        if ( strtolower( $row['DATA_TYPE'] ) != 'text' ) {
            $size = $row['CHARACTER_MAXIMUM_LENGTH'];
        }

        return $size;
    }

    public function limitResult( $sql, $limit, $offset = false ) {
        if ( $offset === false || $offset == 0 ) {
            if ( strpos( $sql, "SELECT" ) === false ) {
                return "TOP {$limit} " . $sql;
            } else {
                return preg_replace( '/\bSELECT(\s+DISTINCT)?\b/Dsi',
                    'SELECT$1 TOP ' . $limit, $sql, 1 );
            }
        } else {
            // This one is fun, we need to pull out the select list as well as any ORDER BY clause
            $select = $orderby = [];
            $s1 = preg_match( '#SELECT\s+(.+?)\s+FROM#Dis', $sql, $select );
            $s2 = preg_match( '#(ORDER BY\s+.+?)(\s*FOR XML .*)?$#Dis', $sql, $orderby );
            $overOrder = $postOrder = '';
            $first = $offset + 1;
            $last = $offset + $limit;
            $sub1 = 'sub_' . $this->mSubqueryId;
            $sub2 = 'sub_' . ( $this->mSubqueryId + 1 );
            $this->mSubqueryId += 2;
            if ( !$s1 ) {
                // wat
                throw new DBUnexpectedError( $this, "Attempting to LIMIT a non-SELECT query\n" );
            }
            if ( !$s2 ) {
                // no ORDER BY
                $overOrder = 'ORDER BY (SELECT 1)';
            } else {
                if ( !isset( $orderby[2] ) || !$orderby[2] ) {
                    // don't need to strip it out if we're using a FOR XML clause
                    $sql = str_replace( $orderby[1], '', $sql );
                }
                $overOrder = $orderby[1];
                $postOrder = ' ' . $overOrder;
            }
            $sql = "SELECT {$select[1]}
                    FROM (
                        SELECT ROW_NUMBER() OVER({$overOrder}) AS rowNumber, *
                        FROM ({$sql}) {$sub1}
                    ) {$sub2}
                    WHERE rowNumber BETWEEN {$first} AND {$last}{$postOrder}";

            return $sql;
        }
    }

    public function LimitToTopN( $sql ) {
        // Matches: LIMIT {[offset,] row_count | row_count OFFSET offset}
        $pattern = '/\bLIMIT\s+((([0-9]+)\s*,\s*)?([0-9]+)(\s+OFFSET\s+([0-9]+))?)/i';
        if ( preg_match( $pattern, $sql, $matches ) ) {
            $row_count = $matches[4];
            $offset = $matches[3] ?: $matches[6] ?: false;

            // strip the matching LIMIT clause out
            $sql = str_replace( $matches[0], '', $sql );

            return $this->limitResult( $sql, $row_count, $offset );
        }

        return $sql;
    }

    public function getSoftwareLink() {
        return "[{{int:version-db-mssql-url}} MS SQL Server]";
    }

    public function getServerVersion() {
        $server_info = sqlsrv_server_info( $this->mConn );
        $version = 'Error';
        if ( isset( $server_info['SQLServerVersion'] ) ) {
            $version = $server_info['SQLServerVersion'];
        }

        return $version;
    }

    public function tableExists( $table, $fname = __METHOD__ ) {
        list( $db, $schema, $table ) = $this->tableName( $table, 'split' );

        if ( $db !== false ) {
            // remote database
            wfDebug( "Attempting to call tableExists on a remote table" );
            return false;
        }

        if ( $schema === false ) {
            global $wgDBmwschema;
            $schema = $wgDBmwschema;
        }

        $res = $this->query( "SELECT 1 FROM INFORMATION_SCHEMA.TABLES
            WHERE TABLE_TYPE = 'BASE TABLE'
            AND TABLE_SCHEMA = '$schema' AND TABLE_NAME = '$table'" );

        if ( $res->numRows() ) {
            return true;
        } else {
            return false;
        }
    }

    public function fieldExists( $table, $field, $fname = __METHOD__ ) {
        list( $db, $schema, $table ) = $this->tableName( $table, 'split' );

        if ( $db !== false ) {
            // remote database
            wfDebug( "Attempting to call fieldExists on a remote table" );
            return false;
        }

        $res = $this->query( "SELECT 1 FROM INFORMATION_SCHEMA.COLUMNS
            WHERE TABLE_SCHEMA = '$schema' AND TABLE_NAME = '$table' AND COLUMN_NAME = '$field'" );

        if ( $res->numRows() ) {
            return true;
        } else {
            return false;
        }
    }

    public function fieldInfo( $table, $field ) {
        list( $db, $schema, $table ) = $this->tableName( $table, 'split' );

        if ( $db !== false ) {
            // remote database
            wfDebug( "Attempting to call fieldInfo on a remote table" );
            return false;
        }

        $res = $this->query( "SELECT * FROM INFORMATION_SCHEMA.COLUMNS
            WHERE TABLE_SCHEMA = '$schema' AND TABLE_NAME = '$table' AND COLUMN_NAME = '$field'" );

        $meta = $res->fetchRow();
        if ( $meta ) {
            return new MssqlField( $meta );
        }

        return false;
    }

    protected function doBegin( $fname = __METHOD__ ) {
        sqlsrv_begin_transaction( $this->mConn );
        $this->mTrxLevel = 1;
    }

    protected function doCommit( $fname = __METHOD__ ) {
        sqlsrv_commit( $this->mConn );
        $this->mTrxLevel = 0;
    }

    protected function doRollback( $fname = __METHOD__ ) {
        sqlsrv_rollback( $this->mConn );
        $this->mTrxLevel = 0;
    }

    private function escapeIdentifier( $identifier ) {
        if ( strlen( $identifier ) == 0 ) {
            throw new InvalidArgumentException( "An identifier must not be empty" );
        }
        if ( strlen( $identifier ) > 128 ) {
            throw new InvalidArgumentException( "The identifier '$identifier' is too long (max. 128)" );
        }
        if ( ( strpos( $identifier, '[' ) !== false )
            || ( strpos( $identifier, ']' ) !== false )
        ) {
            // It may be allowed if you quoted with double quotation marks, but
            // that would break if QUOTED_IDENTIFIER is OFF
            throw new InvalidArgumentException( "Square brackets are not allowed in '$identifier'" );
        }

        return "[$identifier]";
    }

    public function strencode( $s ) {
        // Should not be called by us

        return str_replace( "'", "''", $s );
    }

    public function addQuotes( $s ) {
        if ( $s instanceof MssqlBlob ) {
            return $s->fetch();
        } elseif ( $s instanceof Blob ) {
            // this shouldn't really ever be called, but it's here if needed
            // (and will quite possibly make the SQL error out)
            $blob = new MssqlBlob( $s->fetch() );
            return $blob->fetch();
        } else {
            if ( is_bool( $s ) ) {
                $s = $s ? 1 : 0;
            }
            return parent::addQuotes( $s );
        }
    }

    public function addIdentifierQuotes( $s ) {
        // http://msdn.microsoft.com/en-us/library/aa223962.aspx
        return '[' . $s . ']';
    }

    public function isQuotedIdentifier( $name ) {
        return strlen( $name ) && $name[0] == '[' && substr( $name, -1, 1 ) == ']';
    }

    protected function escapeLikeInternal( $s ) {
        return addcslashes( $s, '\%_[]^' );
    }

    public function buildLike() {
        $params = func_get_args();
        if ( count( $params ) > 0 && is_array( $params[0] ) ) {
            $params = $params[0];
        }

        return parent::buildLike( $params ) . " ESCAPE '\' ";
    }

    public function selectDB( $db ) {
        try {
            $this->mDBname = $db;
            $this->query( "USE $db" );
            return true;
        } catch ( Exception $e ) {
            return false;
        }
    }

    public function makeSelectOptions( $options ) {
        $tailOpts = '';
        $startOpts = '';

        $noKeyOptions = [];
        foreach ( $options as $key => $option ) {
            if ( is_numeric( $key ) ) {
                $noKeyOptions[$option] = true;
            }
        }

        $tailOpts .= $this->makeGroupByWithHaving( $options );

        $tailOpts .= $this->makeOrderBy( $options );

        if ( isset( $noKeyOptions['DISTINCT'] ) || isset( $noKeyOptions['DISTINCTROW'] ) ) {
            $startOpts .= 'DISTINCT';
        }

        if ( isset( $noKeyOptions['FOR XML'] ) ) {
            // used in group concat field emulation
            $tailOpts .= " FOR XML PATH('')";
        }

        // we want this to be compatible with the output of parent::makeSelectOptions()
        return [ $startOpts, '', $tailOpts, '', '' ];
    }

    public function getType() {
        return 'mssql';
    }

    public function buildConcat( $stringList ) {
        return implode( ' + ', $stringList );
    }

    public function buildGroupConcatField( $delim, $table, $field, $conds = '',
        $join_conds = []
    ) {
        $gcsq = 'gcsq_' . $this->mSubqueryId;
        $this->mSubqueryId++;

        $delimLen = strlen( $delim );
        $fld = "{$field} + {$this->addQuotes( $delim )}";
        $sql = "(SELECT LEFT({$field}, LEN({$field}) - {$delimLen}) FROM ("
            . $this->selectSQLText( $table, $fld, $conds, null, [ 'FOR XML' ], $join_conds )
            . ") {$gcsq} ({$field}))";

        return $sql;
    }

    private function getBinaryColumns( $table ) {
        $tableRawArr = explode( '.', preg_replace( '#\[([^\]]*)\]#', '$1', $table ) );
        $tableRaw = array_pop( $tableRawArr );

        if ( $this->mBinaryColumnCache === null ) {
            $this->populateColumnCaches();
        }

        return isset( $this->mBinaryColumnCache[$tableRaw] )
            ? $this->mBinaryColumnCache[$tableRaw]
            : [];
    }

    private function getBitColumns( $table ) {
        $tableRawArr = explode( '.', preg_replace( '#\[([^\]]*)\]#', '$1', $table ) );
        $tableRaw = array_pop( $tableRawArr );

        if ( $this->mBitColumnCache === null ) {
            $this->populateColumnCaches();
        }

        return isset( $this->mBitColumnCache[$tableRaw] )
            ? $this->mBitColumnCache[$tableRaw]
            : [];
    }

    private function populateColumnCaches() {
        $res = $this->select( 'INFORMATION_SCHEMA.COLUMNS', '*',
            [
                'TABLE_CATALOG' => $this->mDBname,
                'TABLE_SCHEMA' => $this->mSchema,
                'DATA_TYPE' => [ 'varbinary', 'binary', 'image', 'bit' ]
            ] );

        $this->mBinaryColumnCache = [];
        $this->mBitColumnCache = [];
        foreach ( $res as $row ) {
            if ( $row->DATA_TYPE == 'bit' ) {
                $this->mBitColumnCache[$row->TABLE_NAME][$row->COLUMN_NAME] = $row;
            } else {
                $this->mBinaryColumnCache[$row->TABLE_NAME][$row->COLUMN_NAME] = $row;
            }
        }
    }

    function tableName( $name, $format = 'quoted' ) {
        # Replace reserved words with better ones
        switch ( $name ) {
            case 'user':
                return $this->realTableName( 'mwuser', $format );
            default:
                return $this->realTableName( $name, $format );
        }
    }

    function realTableName( $name, $format = 'quoted' ) {
        $table = parent::tableName( $name, $format );
        if ( $format == 'split' ) {
            // Used internally, we want the schema split off from the table name and returned
            // as a list with 3 elements (database, schema, table)
            $table = explode( '.', $table );
            while ( count( $table ) < 3 ) {
                array_unshift( $table, false );
            }
        }
        return $table;
    }

    public function dropTable( $tableName, $fName = __METHOD__ ) {
        if ( !$this->tableExists( $tableName, $fName ) ) {
            return false;
        }

        // parent function incorrectly appends CASCADE, which we don't want
        $sql = "DROP TABLE " . $this->tableName( $tableName );

        return $this->query( $sql, $fName );
    }

    public function prepareStatements( $value = null ) {
        return wfSetVar( $this->mPrepareStatements, $value );
    }

    public function scrollableCursor( $value = null ) {
        return wfSetVar( $this->mScrollableCursor, $value );
    }

    public function ignoreErrors( array $value = null ) {
        return wfSetVar( $this->mIgnoreErrors, $value );
    }
} // end DatabaseMssql class