request.js 18 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642
  1. 'use strict'
  2. const msnodesql = require('msnodesqlv8')
  3. const debug = require('debug')('mssql:msv8')
  4. const BaseRequest = require('../base/request')
  5. const RequestError = require('../error/request-error')
  6. const { IDS, objectHasProperty } = require('../utils')
  7. const { TYPES, DECLARATIONS, declare } = require('../datatypes')
  8. const { PARSERS: UDT } = require('../udt')
  9. const Table = require('../table')
  10. const JSON_COLUMN_ID = 'JSON_F52E2B61-18A1-11d1-B105-00805F49916B'
  11. const XML_COLUMN_ID = 'XML_F52E2B61-18A1-11d1-B105-00805F49916B'
  12. const EMPTY_BUFFER = Buffer.alloc(0)
  13. const castParameter = function (value, type) {
  14. if (value == null) {
  15. if ((type === TYPES.Binary) || (type === TYPES.VarBinary) || (type === TYPES.Image)) {
  16. // msnodesql has some problems with NULL values in those types, so we need to replace it with empty buffer
  17. return EMPTY_BUFFER
  18. }
  19. return null
  20. }
  21. switch (type) {
  22. case TYPES.VarChar:
  23. case TYPES.NVarChar:
  24. case TYPES.Char:
  25. case TYPES.NChar:
  26. case TYPES.Xml:
  27. case TYPES.Text:
  28. case TYPES.NText:
  29. if ((typeof value !== 'string') && !(value instanceof String)) {
  30. value = value.toString()
  31. }
  32. break
  33. case TYPES.Int:
  34. case TYPES.TinyInt:
  35. case TYPES.BigInt:
  36. case TYPES.SmallInt:
  37. if ((typeof value !== 'number') && !(value instanceof Number)) {
  38. value = parseInt(value)
  39. if (isNaN(value)) { value = null }
  40. }
  41. break
  42. case TYPES.Float:
  43. case TYPES.Real:
  44. case TYPES.Decimal:
  45. case TYPES.Numeric:
  46. case TYPES.SmallMoney:
  47. case TYPES.Money:
  48. if ((typeof value !== 'number') && !(value instanceof Number)) {
  49. value = parseFloat(value)
  50. if (isNaN(value)) { value = null }
  51. }
  52. break
  53. case TYPES.Bit:
  54. if ((typeof value !== 'boolean') && !(value instanceof Boolean)) {
  55. value = Boolean(value)
  56. }
  57. break
  58. case TYPES.DateTime:
  59. case TYPES.SmallDateTime:
  60. case TYPES.DateTimeOffset:
  61. case TYPES.Date:
  62. if (!(value instanceof Date)) {
  63. value = new Date(value)
  64. }
  65. break
  66. case TYPES.Binary:
  67. case TYPES.VarBinary:
  68. case TYPES.Image:
  69. if (!(value instanceof Buffer)) {
  70. value = Buffer.from(value.toString())
  71. }
  72. break
  73. case TYPES.TVP:
  74. value = msnodesql.TvpFromTable(value)
  75. break
  76. }
  77. return value
  78. }
  79. const createColumns = function (metadata) {
  80. const out = {}
  81. for (let index = 0, length = metadata.length; index < length; index++) {
  82. const column = metadata[index]
  83. out[column.name] = {
  84. index,
  85. name: column.name,
  86. length: column.size,
  87. type: DECLARATIONS[column.sqlType]
  88. }
  89. if (column.udtType != null) {
  90. out[column.name].udt = {
  91. name: column.udtType
  92. }
  93. if (DECLARATIONS[column.udtType]) {
  94. out[column.name].type = DECLARATIONS[column.udtType]
  95. }
  96. }
  97. }
  98. return out
  99. }
  100. const valueCorrection = function (value, metadata) {
  101. if ((metadata.sqlType === 'time') && (value != null)) {
  102. value.setFullYear(1970)
  103. return value
  104. } else if ((metadata.sqlType === 'udt') && (value != null)) {
  105. if (UDT[metadata.udtType]) {
  106. return UDT[metadata.udtType](value)
  107. } else {
  108. return value
  109. }
  110. } else {
  111. return value
  112. }
  113. }
  114. class Request extends BaseRequest {
  115. _batch (batch, callback) {
  116. this._isBatch = true
  117. this._query(batch, callback)
  118. }
  119. _bulk (table, options, callback) {
  120. super._bulk(table, options, err => {
  121. if (err) return callback(err)
  122. table._makeBulk()
  123. if (!table.name) {
  124. setImmediate(callback, new RequestError('Table name must be specified for bulk insert.', 'ENAME'))
  125. }
  126. if (table.name.charAt(0) === '@') {
  127. setImmediate(callback, new RequestError("You can't use table variables for bulk insert.", 'ENAME'))
  128. }
  129. this.parent.acquire(this, (err, connection) => {
  130. let hasReturned = false
  131. if (!err) {
  132. debug('connection(%d): borrowed to request #%d', IDS.get(connection), IDS.get(this))
  133. if (this.canceled) {
  134. debug('request(%d): canceled', IDS.get(this))
  135. this.parent.release(connection)
  136. return callback(new RequestError('Canceled.', 'ECANCEL'))
  137. }
  138. const done = (err, rowCount) => {
  139. if (hasReturned) {
  140. return
  141. }
  142. hasReturned = true
  143. if (err) {
  144. if ((typeof err.sqlstate === 'string') && (err.sqlstate.toLowerCase() === '08s01')) {
  145. connection.hasError = true
  146. }
  147. err = new RequestError(err)
  148. err.code = 'EREQUEST'
  149. }
  150. this.parent.release(connection)
  151. if (err) {
  152. callback(err)
  153. } else {
  154. callback(null, table.rows.length)
  155. }
  156. }
  157. const go = () => {
  158. const tm = connection.tableMgr()
  159. return tm.bind(table.path.replace(/\[|\]/g, ''), mgr => {
  160. if (mgr.columns.length === 0) {
  161. return done(new RequestError('Table was not found on the server.', 'ENAME'))
  162. }
  163. const rows = []
  164. for (const row of Array.from(table.rows)) {
  165. const item = {}
  166. for (let index = 0; index < table.columns.length; index++) {
  167. const col = table.columns[index]
  168. item[col.name] = row[index]
  169. }
  170. rows.push(item)
  171. }
  172. mgr.insertRows(rows, done)
  173. })
  174. }
  175. if (table.create) {
  176. let objectid
  177. if (table.temporary) {
  178. objectid = `tempdb..[${table.name}]`
  179. } else {
  180. objectid = table.path
  181. }
  182. return connection.queryRaw(`if object_id('${objectid.replace(/'/g, '\'\'')}') is null ${table.declare()}`, function (err) {
  183. if (err) { return done(err) }
  184. go()
  185. })
  186. } else {
  187. go()
  188. }
  189. }
  190. })
  191. })
  192. }
  193. _query (command, callback) {
  194. super._query(command, err => {
  195. if (err) return callback(err)
  196. if (command.length === 0) {
  197. return callback(null, [])
  198. }
  199. let row = null
  200. let columns = null
  201. let recordset = null
  202. const recordsets = []
  203. const output = {}
  204. const rowsAffected = []
  205. let handleOutput = false
  206. let isChunkedRecordset = false
  207. let chunksBuffer = null
  208. // nested = function is called by this.execute
  209. if (!this._nested) {
  210. const input = []
  211. for (const name in this.parameters) {
  212. if (!objectHasProperty(this.parameters, name)) {
  213. continue
  214. }
  215. const param = this.parameters[name]
  216. input.push(`@${param.name} ${declare(param.type, param)}`)
  217. }
  218. const sets = []
  219. for (const name in this.parameters) {
  220. if (!objectHasProperty(this.parameters, name)) {
  221. continue
  222. }
  223. const param = this.parameters[name]
  224. if (param.io === 1) {
  225. sets.push(`set @${param.name}=?`)
  226. }
  227. }
  228. const output = []
  229. for (const name in this.parameters) {
  230. if (!objectHasProperty(this.parameters, name)) {
  231. continue
  232. }
  233. const param = this.parameters[name]
  234. if (param.io === 2) {
  235. output.push(`@${param.name} as '${param.name}'`)
  236. }
  237. }
  238. if (input.length) command = `declare ${input.join(',')};${sets.join(';')};${command};`
  239. if (output.length) {
  240. command += `select ${output.join(',')};`
  241. handleOutput = true
  242. }
  243. }
  244. this.parent.acquire(this, (err, connection, config) => {
  245. if (err) return callback(err)
  246. let hasReturned = false
  247. debug('connection(%d): borrowed to request #%d', IDS.get(connection), IDS.get(this))
  248. if (this.canceled) {
  249. debug('request(%d): canceled', IDS.get(this))
  250. this.parent.release(connection)
  251. return callback(new RequestError('Canceled.', 'ECANCEL'))
  252. }
  253. const params = []
  254. for (const name in this.parameters) {
  255. if (!objectHasProperty(this.parameters, name)) {
  256. continue
  257. }
  258. const param = this.parameters[name]
  259. if (param.io === 1) {
  260. params.push(castParameter(param.value, param.type, param))
  261. }
  262. }
  263. debug('request(%d): query', IDS.get(this), command)
  264. const req = connection.queryRaw({
  265. query_str: command,
  266. query_timeout: config.requestTimeout / 1000 // msnodesqlv8 timeouts are in seconds (<1 second not supported)
  267. }, params)
  268. this._setCurrentRequest(req)
  269. this._cancel = () => {
  270. debug('request(%d): cancel', IDS.get(this))
  271. req.cancelQuery(err => {
  272. if (err) debug('request(%d): failed to cancel', IDS.get(this), err)
  273. })
  274. }
  275. req.on('meta', metadata => {
  276. if (row) {
  277. if (isChunkedRecordset) {
  278. const concatenatedChunks = chunksBuffer.join('')
  279. if ((columns[0].name === JSON_COLUMN_ID) && (config.parseJSON === true)) {
  280. try {
  281. if (concatenatedChunks === '') {
  282. row = null
  283. } else {
  284. row = JSON.parse(concatenatedChunks)
  285. }
  286. if (!this.stream) { recordsets[recordsets.length - 1][0] = row }
  287. } catch (ex) {
  288. row = null
  289. const ex2 = new RequestError(`Failed to parse incoming JSON. ${ex.message}`, 'EJSON')
  290. if (this.stream) {
  291. this.emit('error', ex2)
  292. } else {
  293. console.error(ex2)
  294. }
  295. }
  296. } else {
  297. row[columns[0].name] = concatenatedChunks
  298. }
  299. chunksBuffer = null
  300. }
  301. if (row && row.___return___ == null) {
  302. // row with ___return___ col is the last row
  303. if (this.stream) this.emit('row', row)
  304. }
  305. }
  306. row = null
  307. columns = metadata
  308. recordset = []
  309. Object.defineProperty(recordset, 'columns', {
  310. enumerable: false,
  311. configurable: true,
  312. value: createColumns(metadata)
  313. })
  314. Object.defineProperty(recordset, 'toTable', {
  315. enumerable: false,
  316. configurable: true,
  317. value (name) { return Table.fromRecordset(this, name) }
  318. })
  319. isChunkedRecordset = false
  320. if ((metadata.length === 1) && (metadata[0].name === JSON_COLUMN_ID || metadata[0].name === XML_COLUMN_ID)) {
  321. isChunkedRecordset = true
  322. chunksBuffer = []
  323. }
  324. if (this.stream) {
  325. if (recordset.columns.___return___ == null) {
  326. this.emit('recordset', recordset.columns)
  327. }
  328. } else {
  329. recordsets.push(recordset)
  330. }
  331. })
  332. req.on('row', rownumber => {
  333. if (row) {
  334. if (isChunkedRecordset) return
  335. if (row.___return___ == null) {
  336. // row with ___return___ col is the last row
  337. if (this.stream) this.emit('row', row)
  338. }
  339. }
  340. row = {}
  341. if (!this.stream) recordset.push(row)
  342. })
  343. req.on('column', (idx, data, more) => {
  344. if (isChunkedRecordset) {
  345. chunksBuffer.push(data)
  346. } else {
  347. data = valueCorrection(data, columns[idx])
  348. const exi = row[columns[idx].name]
  349. if (exi != null) {
  350. if (exi instanceof Array) {
  351. exi.push(data)
  352. } else {
  353. row[columns[idx].name] = [exi, data]
  354. }
  355. } else {
  356. row[columns[idx].name] = data
  357. }
  358. }
  359. })
  360. req.on('rowcount', count => {
  361. rowsAffected.push(count)
  362. })
  363. req.on('info', msg => {
  364. if ((/^\[Microsoft\]\[SQL Server Native Client 11\.0\](?:\[SQL Server\])?([\s\S]*)$/).exec(msg.message)) {
  365. msg.message = RegExp.$1
  366. }
  367. this.emit('info', {
  368. message: msg.message,
  369. number: msg.code,
  370. state: msg.sqlstate,
  371. class: msg.class || 0,
  372. lineNumber: msg.lineNumber || 0,
  373. serverName: msg.serverName,
  374. procName: msg.procName
  375. })
  376. })
  377. req.once('error', err => {
  378. if (hasReturned) {
  379. return
  380. }
  381. hasReturned = true
  382. if ((typeof err.sqlstate === 'string') && (err.sqlstate.toLowerCase() === '08s01')) {
  383. connection.hasError = true
  384. }
  385. err = new RequestError(err)
  386. err.code = 'EREQUEST'
  387. err.state = err.sqlstate
  388. delete this._cancel
  389. this.parent.release(connection)
  390. debug('request(%d): failed', IDS.get(this), err)
  391. callback(err)
  392. })
  393. req.once('done', () => {
  394. if (hasReturned) {
  395. return
  396. }
  397. hasReturned = true
  398. if (!this._nested) {
  399. if (row) {
  400. if (isChunkedRecordset) {
  401. const concatenatedChunks = chunksBuffer.join('')
  402. if ((columns[0].name === JSON_COLUMN_ID) && (config.parseJSON === true)) {
  403. try {
  404. if (concatenatedChunks === '') {
  405. row = null
  406. } else {
  407. row = JSON.parse(concatenatedChunks)
  408. }
  409. if (!this.stream) { recordsets[recordsets.length - 1][0] = row }
  410. } catch (ex) {
  411. row = null
  412. const ex2 = new RequestError(`Failed to parse incoming JSON. ${ex.message}`, 'EJSON')
  413. if (this.stream) {
  414. this.emit('error', ex2)
  415. } else {
  416. console.error(ex2)
  417. }
  418. }
  419. } else {
  420. row[columns[0].name] = concatenatedChunks
  421. }
  422. chunksBuffer = null
  423. }
  424. if (row && row.___return___ == null) {
  425. // row with ___return___ col is the last row
  426. if (this.stream) { this.emit('row', row) }
  427. }
  428. }
  429. // do we have output parameters to handle?
  430. if (handleOutput && recordsets.length) {
  431. const last = recordsets.pop()[0]
  432. for (const name in this.parameters) {
  433. if (!objectHasProperty(this.parameters, name)) {
  434. continue
  435. }
  436. const param = this.parameters[name]
  437. if (param.io === 2) {
  438. output[param.name] = last[param.name]
  439. }
  440. }
  441. }
  442. }
  443. delete this._cancel
  444. this.parent.release(connection)
  445. debug('request(%d): completed', IDS.get(this))
  446. if (this.stream) {
  447. callback(null, this._nested ? row : null, output, rowsAffected)
  448. } else {
  449. callback(null, recordsets, output, rowsAffected)
  450. }
  451. })
  452. })
  453. })
  454. }
  455. _execute (procedure, callback) {
  456. super._execute(procedure, err => {
  457. if (err) return callback(err)
  458. const params = []
  459. for (const name in this.parameters) {
  460. if (!objectHasProperty(this.parameters, name)) {
  461. continue
  462. }
  463. const param = this.parameters[name]
  464. if (param.io === 2) {
  465. params.push(`@${param.name} ${declare(param.type, param)}`)
  466. }
  467. }
  468. let cmd = `declare ${['@___return___ int'].concat(params).join(', ')};`
  469. cmd += `exec @___return___ = ${procedure} `
  470. const spp = []
  471. for (const name in this.parameters) {
  472. if (!objectHasProperty(this.parameters, name)) {
  473. continue
  474. }
  475. const param = this.parameters[name]
  476. if (param.io === 2) {
  477. // output parameter
  478. spp.push(`@${param.name}=@${param.name} output`)
  479. } else {
  480. // input parameter
  481. spp.push(`@${param.name}=?`)
  482. }
  483. }
  484. const params2 = []
  485. for (const name in this.parameters) {
  486. if (!objectHasProperty(this.parameters, name)) {
  487. continue
  488. }
  489. const param = this.parameters[name]
  490. if (param.io === 2) {
  491. params2.push(`@${param.name} as '${param.name}'`)
  492. }
  493. }
  494. cmd += `${spp.join(', ')};`
  495. cmd += `select ${['@___return___ as \'___return___\''].concat(params2).join(', ')};`
  496. this._nested = true
  497. this._query(cmd, (err, recordsets, output, rowsAffected) => {
  498. this._nested = false
  499. if (err) return callback(err)
  500. let last, returnValue
  501. if (this.stream) {
  502. last = recordsets
  503. } else {
  504. last = recordsets.pop()
  505. if (last) last = last[0]
  506. }
  507. if (last && (last.___return___ != null)) {
  508. returnValue = last.___return___
  509. for (const name in this.parameters) {
  510. if (!objectHasProperty(this.parameters, name)) {
  511. continue
  512. }
  513. const param = this.parameters[name]
  514. if (param.io === 2) {
  515. output[param.name] = last[param.name]
  516. }
  517. }
  518. }
  519. if (this.stream) {
  520. callback(null, null, output, returnValue, rowsAffected)
  521. } else {
  522. callback(null, recordsets, output, returnValue, rowsAffected)
  523. }
  524. })
  525. })
  526. }
  527. _pause () {
  528. super._pause()
  529. if (this._currentRequest) {
  530. this._currentRequest.pauseQuery()
  531. }
  532. }
  533. _resume () {
  534. super._resume()
  535. if (this._currentRequest) {
  536. this._currentRequest.resumeQuery()
  537. }
  538. }
  539. }
  540. module.exports = Request