DatabaseMssql.php

Go to the documentation of this file.

<?php
namespace Wikimedia\Rdbms;
 
use Wikimedia;
use Exception;
use stdClass;
 
class DatabaseMssql extends Database {
    protected $serverPort;
    protected $useWindowsAuth = false;
    protected $lastInsertId = null;
    protected $lastAffectedRowCount = null;
    protected $subqueryId = 0;
    protected $scrollableCursor = true;
    protected $prepareStatements = true;
    protected $binaryColumnCache = null;
    protected $bitColumnCache = null;
    protected $ignoreDupKeyErrors = false;
    protected $ignoreErrors = [];
 
    public function implicitGroupby() {
        return false;
    }
 
    public function implicitOrderby() {
        return false;
    }
 
    public function unionSupportsOrderAndLimit() {
        return false;
    }
 
    public function __construct( array $params ) {
        $this->serverPort = $params['port'];
        $this->useWindowsAuth = $params['UseWindowsAuth'];
 
        parent::__construct( $params );
    }
 
    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"
            );
        }
 
        # e.g. the class is being loaded
        if ( !strlen( $user ) ) {
            return null;
        }
 
        $this->close();
        $this->server = $server;
        $this->user = $user;
        $this->password = $password;
        $this->dbName = $dbName;
 
        $connectionInfo = [];
 
        if ( $dbName ) {
            $connectionInfo['Database'] = $dbName;
        }
 
        // Decide which auth scenerio to use
        // if we are using Windows auth, then don't add credentials to $connectionInfo
        if ( !$this->useWindowsAuth ) {
            $connectionInfo['UID'] = $user;
            $connectionInfo['PWD'] = $password;
        }
 
        Wikimedia\suppressWarnings();
        $this->conn = sqlsrv_connect( $server, $connectionInfo );
        Wikimedia\restoreWarnings();
 
        if ( $this->conn === false ) {
            throw new DBConnectionError( $this, $this->lastError() );
        }
 
        $this->opened = true;
 
        return $this->conn;
    }
 
    protected function closeConnection() {
        return sqlsrv_close( $this->conn );
    }
 
    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 ) {
        // several extensions seem to think that all databases support limits
        // via LIMIT N after the WHERE clause, but  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->scrollableCursor ) {
            $scrollArr = [ 'Scrollable' => SQLSRV_CURSOR_STATIC ];
        } else {
            $scrollArr = [];
        }
 
        if ( $this->prepareStatements ) {
            // we do prepare + execute so we can get its field metadata for later usage if desired
            $stmt = sqlsrv_prepare( $this->conn, $sql, [], $scrollArr );
            $success = sqlsrv_execute( $stmt );
        } else {
            $stmt = sqlsrv_query( $this->conn, $sql, [], $scrollArr );
            $success = (bool)$stmt;
        }
 
        // Make a copy to ensure what we add below does not get reflected in future queries
        $ignoreErrors = $this->ignoreErrors;
 
        if ( $this->ignoreDupKeyErrors ) {
            // 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->lastAffectedRowCount = 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->lastInsertId;
    }
 
    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;
        }
    }
 
    protected function wasKnownStatementRollbackError() {
        $errors = sqlsrv_errors( SQLSRV_ERR_ALL );
        if ( !$errors ) {
            return false;
        }
        // The transaction vs statement rollback behavior depends on XACT_ABORT, so make sure
        // that the "statement has been terminated" error (3621) is specifically present.
        // https://docs.microsoft.com/en-us/sql/t-sql/statements/set-xact-abort-transact-sql
        $statementOnly = false;
        $codeWhitelist = [ '2601', '2627', '547' ];
        foreach ( $errors as $error ) {
            if ( $error['code'] == '3621' ) {
                $statementOnly = true;
            } elseif ( !in_array( $error['code'], $codeWhitelist ) ) {
                $statementOnly = false;
                break;
            }
        }
 
        return $statementOnly;
    }
 
    protected function fetchAffectedRowCount() {
        return $this->lastAffectedRowCount;
    }
 
    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->scrollableCursor = false;
                $this->prepareStatements = 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->scrollableCursor = true;
                    $this->prepareStatements = true;
                    throw $dqe;
                }
            }
            $this->scrollableCursor = true;
            $this->prepareStatements = 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 ) ) {
                $tables = $table;
                while ( $tables ) {
                    $t = array_pop( $tables );
                    if ( is_array( $t ) ) {
                        $tables = array_merge( $tables, $t );
                    } else {
                        $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->scrollableCursor = false;
        try {
            parent::deleteJoin( $delTable, $joinTable, $delVar, $joinVar, $conds, $fname );
        } catch ( Exception $e ) {
            $this->scrollableCursor = true;
            throw $e;
        }
        $this->scrollableCursor = true;
    }
 
    public function delete( $table, $conds, $fname = __METHOD__ ) {
        $this->scrollableCursor = false;
        try {
            parent::delete( $table, $conds, $fname );
        } catch ( Exception $e ) {
            $this->scrollableCursor = true;
            throw $e;
        }
        $this->scrollableCursor = true;
    }
 
    public function estimateRowCount( $table, $var = '*', $conds = '',
        $fname = __METHOD__, $options = [], $join_conds = []
    ) {
        $conds = $this->normalizeConditions( $conds, $fname );
        $column = $this->extractSingleFieldFromList( $var );
        if ( is_string( $column ) && !in_array( $column, [ '*', '1' ] ) ) {
            $conds[] = "$column IS NOT NULL";
        }
 
        // http://msdn2.microsoft.com/en-us/library/aa259203.aspx
        $options['EXPLAIN'] = true;
        $options['FOR COUNT'] = true;
        $res = $this->select( $table, $var, $conds, $fname, $options, $join_conds );
 
        $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 existence 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 $result ?: false;
    }
 
    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->ignoreDupKeyErrors = true;
        }
 
        $ret = null;
        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->scrollableCursor = false;
            try {
                $ret = $this->query( $sql );
            } catch ( Exception $e ) {
                $this->scrollableCursor = true;
                $this->ignoreDupKeyErrors = false;
                throw $e;
            }
            $this->scrollableCursor = true;
 
            if ( $ret instanceof ResultWrapper && !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->lastInsertId = $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->lastAffectedRowCount == -1 ) {
                        $this->lastAffectedRowCount = 1;
                    }
                }
            }
        }
 
        $this->ignoreDupKeyErrors = false;
 
        return $ret;
    }
 
    public function nativeInsertSelect( $destTable, $srcTable, $varMap, $conds, $fname = __METHOD__,
        $insertOptions = [], $selectOptions = [], $selectJoinConds = []
    ) {
        $this->scrollableCursor = false;
        try {
            $ret = parent::nativeInsertSelect(
                $destTable,
                $srcTable,
                $varMap,
                $conds,
                $fname,
                $insertOptions,
                $selectOptions,
                $selectJoinConds
            );
        } catch ( Exception $e ) {
            $this->scrollableCursor = true;
            throw $e;
        }
        $this->scrollableCursor = 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->scrollableCursor = false;
        try {
            $this->query( $sql );
        } catch ( Exception $e ) {
            $this->scrollableCursor = true;
            throw $e;
        }
        $this->scrollableCursor = 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 );
            $postOrder = '';
            $first = $offset + 1;
            $last = $offset + $limit;
            $sub1 = 'sub_' . $this->subqueryId;
            $sub2 = 'sub_' . ( $this->subqueryId + 1 );
            $this->subqueryId += 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->conn );
        $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
            $this->queryLogger->error( "Attempting to call tableExists on a remote table" );
            return false;
        }
 
        if ( $schema === false ) {
            $schema = $this->schema;
        }
 
        $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
            $this->queryLogger->error( "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
            $this->queryLogger->error( "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 doSavepoint( $identifier, $fname ) {
        $this->query( 'SAVE TRANSACTION ' . $this->addIdentifierQuotes( $identifier ), $fname );
    }
 
    protected function doReleaseSavepoint( $identifier, $fname ) {
        // Not supported. Also not really needed, a new doSavepoint() for the
        // same identifier will overwrite the old.
    }
 
    protected function doRollbackToSavepoint( $identifier, $fname ) {
        $this->query( 'ROLLBACK TRANSACTION ' . $this->addIdentifierQuotes( $identifier ), $fname );
    }
 
    protected function doBegin( $fname = __METHOD__ ) {
        sqlsrv_begin_transaction( $this->conn );
        $this->trxLevel = 1;
    }
 
    protected function doCommit( $fname = __METHOD__ ) {
        sqlsrv_commit( $this->conn );
        $this->trxLevel = 0;
    }
 
    protected function doRollback( $fname = __METHOD__ ) {
        sqlsrv_rollback( $this->conn );
        $this->trxLevel = 0;
    }
 
    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, $escapeChar = '`' ) {
        return str_replace( [ $escapeChar, '%', '_', '[', ']', '^' ],
            [ "{$escapeChar}{$escapeChar}", "{$escapeChar}%", "{$escapeChar}_",
                "{$escapeChar}[", "{$escapeChar}]", "{$escapeChar}^" ],
            $s );
    }
 
    public function selectDB( $db ) {
        try {
            $this->dbName = $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->subqueryId;
        $this->subqueryId++;
 
        $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;
    }
 
    public function buildSubstring( $input, $startPosition, $length = null ) {
        $this->assertBuildSubstringParams( $startPosition, $length );
        if ( $length === null ) {
            $length = 2147483647;
        }
        return 'SUBSTRING(' . implode( ',', [ $input, $startPosition, $length ] ) . ')';
    }
 
    private function getBinaryColumns( $table ) {
        $tableRawArr = explode( '.', preg_replace( '#\[([^\]]*)\]#', '$1', $table ) );
        $tableRaw = array_pop( $tableRawArr );
 
        if ( $this->binaryColumnCache === null ) {
            $this->populateColumnCaches();
        }
 
        return isset( $this->binaryColumnCache[$tableRaw] )
            ? $this->binaryColumnCache[$tableRaw]
            : [];
    }
 
    private function getBitColumns( $table ) {
        $tableRawArr = explode( '.', preg_replace( '#\[([^\]]*)\]#', '$1', $table ) );
        $tableRaw = array_pop( $tableRawArr );
 
        if ( $this->bitColumnCache === null ) {
            $this->populateColumnCaches();
        }
 
        return isset( $this->bitColumnCache[$tableRaw] )
            ? $this->bitColumnCache[$tableRaw]
            : [];
    }
 
    private function populateColumnCaches() {
        $res = $this->select( 'INFORMATION_SCHEMA.COLUMNS', '*',
            [
                'TABLE_CATALOG' => $this->dbName,
                'TABLE_SCHEMA' => $this->schema,
                'DATA_TYPE' => [ 'varbinary', 'binary', 'image', 'bit' ]
            ] );
 
        $this->binaryColumnCache = [];
        $this->bitColumnCache = [];
        foreach ( $res as $row ) {
            if ( $row->DATA_TYPE == 'bit' ) {
                $this->bitColumnCache[$row->TABLE_NAME][$row->COLUMN_NAME] = $row;
            } else {
                $this->binaryColumnCache[$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 ) {
        $old = $this->prepareStatements;
        if ( $value !== null ) {
            $this->prepareStatements = $value;
        }
 
        return $old;
    }
 
    public function scrollableCursor( $value = null ) {
        $old = $this->scrollableCursor;
        if ( $value !== null ) {
            $this->scrollableCursor = $value;
        }
 
        return $old;
    }
}
 
class_alias( DatabaseMssql::class, 'DatabaseMssql' );