| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642 |
- 'use strict'
- const msnodesql = require('msnodesqlv8')
- const debug = require('debug')('mssql:msv8')
- const BaseRequest = require('../base/request')
- const RequestError = require('../error/request-error')
- const { IDS, objectHasProperty } = require('../utils')
- const { TYPES, DECLARATIONS, declare } = require('../datatypes')
- const { PARSERS: UDT } = require('../udt')
- const Table = require('../table')
- const JSON_COLUMN_ID = 'JSON_F52E2B61-18A1-11d1-B105-00805F49916B'
- const XML_COLUMN_ID = 'XML_F52E2B61-18A1-11d1-B105-00805F49916B'
- const EMPTY_BUFFER = Buffer.alloc(0)
- const castParameter = function (value, type) {
- if (value == null) {
- if ((type === TYPES.Binary) || (type === TYPES.VarBinary) || (type === TYPES.Image)) {
- // msnodesql has some problems with NULL values in those types, so we need to replace it with empty buffer
- return EMPTY_BUFFER
- }
- return null
- }
- switch (type) {
- case TYPES.VarChar:
- case TYPES.NVarChar:
- case TYPES.Char:
- case TYPES.NChar:
- case TYPES.Xml:
- case TYPES.Text:
- case TYPES.NText:
- if ((typeof value !== 'string') && !(value instanceof String)) {
- value = value.toString()
- }
- break
- case TYPES.Int:
- case TYPES.TinyInt:
- case TYPES.BigInt:
- case TYPES.SmallInt:
- if ((typeof value !== 'number') && !(value instanceof Number)) {
- value = parseInt(value)
- if (isNaN(value)) { value = null }
- }
- break
- case TYPES.Float:
- case TYPES.Real:
- case TYPES.Decimal:
- case TYPES.Numeric:
- case TYPES.SmallMoney:
- case TYPES.Money:
- if ((typeof value !== 'number') && !(value instanceof Number)) {
- value = parseFloat(value)
- if (isNaN(value)) { value = null }
- }
- break
- case TYPES.Bit:
- if ((typeof value !== 'boolean') && !(value instanceof Boolean)) {
- value = Boolean(value)
- }
- break
- case TYPES.DateTime:
- case TYPES.SmallDateTime:
- case TYPES.DateTimeOffset:
- case TYPES.Date:
- if (!(value instanceof Date)) {
- value = new Date(value)
- }
- break
- case TYPES.Binary:
- case TYPES.VarBinary:
- case TYPES.Image:
- if (!(value instanceof Buffer)) {
- value = Buffer.from(value.toString())
- }
- break
- case TYPES.TVP:
- value = msnodesql.TvpFromTable(value)
- break
- }
- return value
- }
- const createColumns = function (metadata) {
- const out = {}
- for (let index = 0, length = metadata.length; index < length; index++) {
- const column = metadata[index]
- out[column.name] = {
- index,
- name: column.name,
- length: column.size,
- type: DECLARATIONS[column.sqlType]
- }
- if (column.udtType != null) {
- out[column.name].udt = {
- name: column.udtType
- }
- if (DECLARATIONS[column.udtType]) {
- out[column.name].type = DECLARATIONS[column.udtType]
- }
- }
- }
- return out
- }
- const valueCorrection = function (value, metadata) {
- if ((metadata.sqlType === 'time') && (value != null)) {
- value.setFullYear(1970)
- return value
- } else if ((metadata.sqlType === 'udt') && (value != null)) {
- if (UDT[metadata.udtType]) {
- return UDT[metadata.udtType](value)
- } else {
- return value
- }
- } else {
- return value
- }
- }
- class Request extends BaseRequest {
- _batch (batch, callback) {
- this._isBatch = true
- this._query(batch, callback)
- }
- _bulk (table, options, callback) {
- super._bulk(table, options, err => {
- if (err) return callback(err)
- table._makeBulk()
- if (!table.name) {
- setImmediate(callback, new RequestError('Table name must be specified for bulk insert.', 'ENAME'))
- }
- if (table.name.charAt(0) === '@') {
- setImmediate(callback, new RequestError("You can't use table variables for bulk insert.", 'ENAME'))
- }
- this.parent.acquire(this, (err, connection) => {
- let hasReturned = false
- if (!err) {
- debug('connection(%d): borrowed to request #%d', IDS.get(connection), IDS.get(this))
- if (this.canceled) {
- debug('request(%d): canceled', IDS.get(this))
- this.parent.release(connection)
- return callback(new RequestError('Canceled.', 'ECANCEL'))
- }
- const done = (err, rowCount) => {
- if (hasReturned) {
- return
- }
- hasReturned = true
- if (err) {
- if ((typeof err.sqlstate === 'string') && (err.sqlstate.toLowerCase() === '08s01')) {
- connection.hasError = true
- }
- err = new RequestError(err)
- err.code = 'EREQUEST'
- }
- this.parent.release(connection)
- if (err) {
- callback(err)
- } else {
- callback(null, table.rows.length)
- }
- }
- const go = () => {
- const tm = connection.tableMgr()
- return tm.bind(table.path.replace(/\[|\]/g, ''), mgr => {
- if (mgr.columns.length === 0) {
- return done(new RequestError('Table was not found on the server.', 'ENAME'))
- }
- const rows = []
- for (const row of Array.from(table.rows)) {
- const item = {}
- for (let index = 0; index < table.columns.length; index++) {
- const col = table.columns[index]
- item[col.name] = row[index]
- }
- rows.push(item)
- }
- mgr.insertRows(rows, done)
- })
- }
- if (table.create) {
- let objectid
- if (table.temporary) {
- objectid = `tempdb..[${table.name}]`
- } else {
- objectid = table.path
- }
- return connection.queryRaw(`if object_id('${objectid.replace(/'/g, '\'\'')}') is null ${table.declare()}`, function (err) {
- if (err) { return done(err) }
- go()
- })
- } else {
- go()
- }
- }
- })
- })
- }
- _query (command, callback) {
- super._query(command, err => {
- if (err) return callback(err)
- if (command.length === 0) {
- return callback(null, [])
- }
- let row = null
- let columns = null
- let recordset = null
- const recordsets = []
- const output = {}
- const rowsAffected = []
- let handleOutput = false
- let isChunkedRecordset = false
- let chunksBuffer = null
- // nested = function is called by this.execute
- if (!this._nested) {
- const input = []
- for (const name in this.parameters) {
- if (!objectHasProperty(this.parameters, name)) {
- continue
- }
- const param = this.parameters[name]
- input.push(`@${param.name} ${declare(param.type, param)}`)
- }
- const sets = []
- for (const name in this.parameters) {
- if (!objectHasProperty(this.parameters, name)) {
- continue
- }
- const param = this.parameters[name]
- if (param.io === 1) {
- sets.push(`set @${param.name}=?`)
- }
- }
- const output = []
- for (const name in this.parameters) {
- if (!objectHasProperty(this.parameters, name)) {
- continue
- }
- const param = this.parameters[name]
- if (param.io === 2) {
- output.push(`@${param.name} as '${param.name}'`)
- }
- }
- if (input.length) command = `declare ${input.join(',')};${sets.join(';')};${command};`
- if (output.length) {
- command += `select ${output.join(',')};`
- handleOutput = true
- }
- }
- this.parent.acquire(this, (err, connection, config) => {
- if (err) return callback(err)
- let hasReturned = false
- debug('connection(%d): borrowed to request #%d', IDS.get(connection), IDS.get(this))
- if (this.canceled) {
- debug('request(%d): canceled', IDS.get(this))
- this.parent.release(connection)
- return callback(new RequestError('Canceled.', 'ECANCEL'))
- }
- const params = []
- for (const name in this.parameters) {
- if (!objectHasProperty(this.parameters, name)) {
- continue
- }
- const param = this.parameters[name]
- if (param.io === 1) {
- params.push(castParameter(param.value, param.type, param))
- }
- }
- debug('request(%d): query', IDS.get(this), command)
- const req = connection.queryRaw({
- query_str: command,
- query_timeout: config.requestTimeout / 1000 // msnodesqlv8 timeouts are in seconds (<1 second not supported)
- }, params)
- this._setCurrentRequest(req)
- this._cancel = () => {
- debug('request(%d): cancel', IDS.get(this))
- req.cancelQuery(err => {
- if (err) debug('request(%d): failed to cancel', IDS.get(this), err)
- })
- }
- req.on('meta', metadata => {
- if (row) {
- if (isChunkedRecordset) {
- const concatenatedChunks = chunksBuffer.join('')
- if ((columns[0].name === JSON_COLUMN_ID) && (config.parseJSON === true)) {
- try {
- if (concatenatedChunks === '') {
- row = null
- } else {
- row = JSON.parse(concatenatedChunks)
- }
- if (!this.stream) { recordsets[recordsets.length - 1][0] = row }
- } catch (ex) {
- row = null
- const ex2 = new RequestError(`Failed to parse incoming JSON. ${ex.message}`, 'EJSON')
- if (this.stream) {
- this.emit('error', ex2)
- } else {
- console.error(ex2)
- }
- }
- } else {
- row[columns[0].name] = concatenatedChunks
- }
- chunksBuffer = null
- }
- if (row && row.___return___ == null) {
- // row with ___return___ col is the last row
- if (this.stream) this.emit('row', row)
- }
- }
- row = null
- columns = metadata
- recordset = []
- Object.defineProperty(recordset, 'columns', {
- enumerable: false,
- configurable: true,
- value: createColumns(metadata)
- })
- Object.defineProperty(recordset, 'toTable', {
- enumerable: false,
- configurable: true,
- value (name) { return Table.fromRecordset(this, name) }
- })
- isChunkedRecordset = false
- if ((metadata.length === 1) && (metadata[0].name === JSON_COLUMN_ID || metadata[0].name === XML_COLUMN_ID)) {
- isChunkedRecordset = true
- chunksBuffer = []
- }
- if (this.stream) {
- if (recordset.columns.___return___ == null) {
- this.emit('recordset', recordset.columns)
- }
- } else {
- recordsets.push(recordset)
- }
- })
- req.on('row', rownumber => {
- if (row) {
- if (isChunkedRecordset) return
- if (row.___return___ == null) {
- // row with ___return___ col is the last row
- if (this.stream) this.emit('row', row)
- }
- }
- row = {}
- if (!this.stream) recordset.push(row)
- })
- req.on('column', (idx, data, more) => {
- if (isChunkedRecordset) {
- chunksBuffer.push(data)
- } else {
- data = valueCorrection(data, columns[idx])
- const exi = row[columns[idx].name]
- if (exi != null) {
- if (exi instanceof Array) {
- exi.push(data)
- } else {
- row[columns[idx].name] = [exi, data]
- }
- } else {
- row[columns[idx].name] = data
- }
- }
- })
- req.on('rowcount', count => {
- rowsAffected.push(count)
- })
- req.on('info', msg => {
- if ((/^\[Microsoft\]\[SQL Server Native Client 11\.0\](?:\[SQL Server\])?([\s\S]*)$/).exec(msg.message)) {
- msg.message = RegExp.$1
- }
- this.emit('info', {
- message: msg.message,
- number: msg.code,
- state: msg.sqlstate,
- class: msg.class || 0,
- lineNumber: msg.lineNumber || 0,
- serverName: msg.serverName,
- procName: msg.procName
- })
- })
- req.once('error', err => {
- if (hasReturned) {
- return
- }
- hasReturned = true
- if ((typeof err.sqlstate === 'string') && (err.sqlstate.toLowerCase() === '08s01')) {
- connection.hasError = true
- }
- err = new RequestError(err)
- err.code = 'EREQUEST'
- err.state = err.sqlstate
- delete this._cancel
- this.parent.release(connection)
- debug('request(%d): failed', IDS.get(this), err)
- callback(err)
- })
- req.once('done', () => {
- if (hasReturned) {
- return
- }
- hasReturned = true
- if (!this._nested) {
- if (row) {
- if (isChunkedRecordset) {
- const concatenatedChunks = chunksBuffer.join('')
- if ((columns[0].name === JSON_COLUMN_ID) && (config.parseJSON === true)) {
- try {
- if (concatenatedChunks === '') {
- row = null
- } else {
- row = JSON.parse(concatenatedChunks)
- }
- if (!this.stream) { recordsets[recordsets.length - 1][0] = row }
- } catch (ex) {
- row = null
- const ex2 = new RequestError(`Failed to parse incoming JSON. ${ex.message}`, 'EJSON')
- if (this.stream) {
- this.emit('error', ex2)
- } else {
- console.error(ex2)
- }
- }
- } else {
- row[columns[0].name] = concatenatedChunks
- }
- chunksBuffer = null
- }
- if (row && row.___return___ == null) {
- // row with ___return___ col is the last row
- if (this.stream) { this.emit('row', row) }
- }
- }
- // do we have output parameters to handle?
- if (handleOutput && recordsets.length) {
- const last = recordsets.pop()[0]
- for (const name in this.parameters) {
- if (!objectHasProperty(this.parameters, name)) {
- continue
- }
- const param = this.parameters[name]
- if (param.io === 2) {
- output[param.name] = last[param.name]
- }
- }
- }
- }
- delete this._cancel
- this.parent.release(connection)
- debug('request(%d): completed', IDS.get(this))
- if (this.stream) {
- callback(null, this._nested ? row : null, output, rowsAffected)
- } else {
- callback(null, recordsets, output, rowsAffected)
- }
- })
- })
- })
- }
- _execute (procedure, callback) {
- super._execute(procedure, err => {
- if (err) return callback(err)
- const params = []
- for (const name in this.parameters) {
- if (!objectHasProperty(this.parameters, name)) {
- continue
- }
- const param = this.parameters[name]
- if (param.io === 2) {
- params.push(`@${param.name} ${declare(param.type, param)}`)
- }
- }
- let cmd = `declare ${['@___return___ int'].concat(params).join(', ')};`
- cmd += `exec @___return___ = ${procedure} `
- const spp = []
- for (const name in this.parameters) {
- if (!objectHasProperty(this.parameters, name)) {
- continue
- }
- const param = this.parameters[name]
- if (param.io === 2) {
- // output parameter
- spp.push(`@${param.name}=@${param.name} output`)
- } else {
- // input parameter
- spp.push(`@${param.name}=?`)
- }
- }
- const params2 = []
- for (const name in this.parameters) {
- if (!objectHasProperty(this.parameters, name)) {
- continue
- }
- const param = this.parameters[name]
- if (param.io === 2) {
- params2.push(`@${param.name} as '${param.name}'`)
- }
- }
- cmd += `${spp.join(', ')};`
- cmd += `select ${['@___return___ as \'___return___\''].concat(params2).join(', ')};`
- this._nested = true
- this._query(cmd, (err, recordsets, output, rowsAffected) => {
- this._nested = false
- if (err) return callback(err)
- let last, returnValue
- if (this.stream) {
- last = recordsets
- } else {
- last = recordsets.pop()
- if (last) last = last[0]
- }
- if (last && (last.___return___ != null)) {
- returnValue = last.___return___
- for (const name in this.parameters) {
- if (!objectHasProperty(this.parameters, name)) {
- continue
- }
- const param = this.parameters[name]
- if (param.io === 2) {
- output[param.name] = last[param.name]
- }
- }
- }
- if (this.stream) {
- callback(null, null, output, returnValue, rowsAffected)
- } else {
- callback(null, recordsets, output, returnValue, rowsAffected)
- }
- })
- })
- }
- _pause () {
- super._pause()
- if (this._currentRequest) {
- this._currentRequest.pauseQuery()
- }
- }
- _resume () {
- super._resume()
- if (this._currentRequest) {
- this._currentRequest.resumeQuery()
- }
- }
- }
- module.exports = Request
|