UPSERT functionality for MSSQL (#6842)

* Added upsertQuery for MSSQL

* Updated changelog and model api

* Add missing space

* Removed parenthesis from arrow functions

* Fixes edge case when two unique keys are present in model, added WITH(HOLDLOCK) to prevent concurrency issues, integration test for multiple unique keys

.gitignore

@@ -14,3 +14,4 @@ site
 docs/api/tmp.md
 ssce.js
 coverage
+.vscode/
\ No newline at end of file

changelog.md

 # Future
+- [ADDED] UPSERT Support for MSSQL [#6842](https://github.com/sequelize/sequelize/pull/6842)
 - [FIXED] Execute queries parallel in findAndCount [#6695](https://github.com/sequelize/sequelize/issues/6695)
 - [FIXED] `restore` now uses `field` from `deletedAt`
 - [FIXED] MSSQL bulkInsertQuery when options and attributes are not passed

lib/dialects/mssql/index.js

@@ -26,7 +26,7 @@ MssqlDialect.prototype.supports = _.merge(_.cloneDeep(Abstract.prototype.support
   lock: false,
   transactions: false,
   migrations: false,
-  upserts: false,
+  upserts: true,
   returnValues: {
     output: true
   },

lib/dialects/mssql/query-generator.js

@@ -302,6 +302,112 @@ var QueryGenerator = {
     return generatedQuery;
   },
 
+  upsertQuery(tableName, insertValues, updateValues, where, rawAttributes, options) {
+    const targetTableAlias = this.quoteTable(`${tableName}_target`);
+    const sourceTableAlias = this.quoteTable(`${tableName}_source`);
+    const primaryKeysAttrs = [];
+    const identityAttrs = [];
+    const uniqueAttrs = [];
+    const tableNameQuoted = this.quoteTable(tableName);
+    let needIdentityInsertWrapper = false;
+
+    
+    //Obtain primaryKeys, uniquekeys and identity attrs from rawAttributes as model is not passed
+    for (const key in rawAttributes) {
+      if (rawAttributes[key].primaryKey) {
+        primaryKeysAttrs.push(rawAttributes[key].field || key);
+      }
+      if (rawAttributes[key].unique) {
+        uniqueAttrs.push(rawAttributes[key].field || key);
+      }
+      if (rawAttributes[key].autoIncrement) {
+        identityAttrs.push(rawAttributes[key].field || key);
+      }
+    }
+
+    const updateKeys = Object.keys(updateValues);
+    const insertKeys = Object.keys(insertValues);
+    const insertKeysQuoted = insertKeys.map(key => this.quoteIdentifier(key)).join(', ');
+    const insertValuesEscaped = insertKeys.map(key => this.escape(insertValues[key])).join(', ');
+    const sourceTableQuery = `VALUES(${insertValuesEscaped})`; //Virtual Table
+    let joinCondition;
+
+    //IDENTITY_INSERT Condition
+    identityAttrs.forEach(key => {
+      if (updateValues[key] && updateValues[key] !== null) {
+        needIdentityInsertWrapper = true;
+        /*
+         * IDENTITY_INSERT Column Cannot be updated, only inserted
+         * http://stackoverflow.com/a/30176254/2254360
+         */
+      }
+    });
+
+    //Filter NULL Clauses
+    const clauses = where.$or.filter(clause => {
+      let valid = true;
+      /*
+       * Exclude NULL Composite PK/UK. Partial Composite clauses should also be excluded as it doesn't guarantee a single row
+       */
+      for (const key in clause) {
+        if (!clause[key]) {
+          valid = false;
+          break;
+        }
+      }
+      return valid;
+    });
+
+    /*
+     * Generate ON condition using PK(s).
+     * If not, generate using UK(s). Else throw error
+     */
+    const getJoinSnippet = array => {
+      return array.map(key => {
+        key = this.quoteIdentifier(key);
+        return `${targetTableAlias}.${key} = ${sourceTableAlias}.${key}`;
+      });
+    };
+
+    if (clauses.length === 0) {
+      throw new Error('Primary Key or Unique key should be passed to upsert query');      
+    } else {
+      // Search for primary key attribute in clauses -- Model can have two separate unique keys
+      for (const key in clauses) {
+        const keys = Object.keys(clauses[key]);
+        if (primaryKeysAttrs.indexOf(keys[0]) !== -1) {
+          joinCondition = getJoinSnippet(primaryKeysAttrs).join(' AND ');
+          break;
+        }
+      }
+      if (!joinCondition) {
+        joinCondition = getJoinSnippet(uniqueAttrs).join(' AND ');
+      }
+    }
+    
+    // Remove the IDENTITY_INSERT Column from update
+    const updateSnippet = updateKeys.filter(key => {
+      if (identityAttrs.indexOf(key) === -1) {
+        return true;
+      } else {
+        return false;
+      }
+    })
+    .map(key => {
+      const value = this.escape(updateValues[key]);
+      key = this.quoteIdentifier(key);
+      return `${targetTableAlias}.${key} = ${value}`;
+    }).join(', ');
+
+    const insertSnippet = `(${insertKeysQuoted}) VALUES(${insertValuesEscaped})`;
+    let query = `MERGE INTO ${tableNameQuoted} WITH(HOLDLOCK) AS ${targetTableAlias} USING (${sourceTableQuery}) AS ${sourceTableAlias}(${insertKeysQuoted}) ON ${joinCondition}`;
+    query += ` WHEN MATCHED THEN UPDATE SET ${updateSnippet} WHEN NOT MATCHED THEN INSERT ${insertSnippet} OUTPUT $action, INSERTED.*;`;
+    if (needIdentityInsertWrapper) {
+      query = `SET IDENTITY_INSERT ${tableNameQuoted} ON; ${query} SET IDENTITY_INSERT ${tableNameQuoted} OFF;`;
+    }
+    return query;
+  },
+
   deleteQuery(tableName, where, options) {
     options = options || {};
 

lib/dialects/mssql/query.js

@@ -169,6 +169,13 @@ class Query extends AbstractQuery {
       result = this.handleShowIndexesQuery(data);
     } else if (this.isSelectQuery()) {
       result = this.handleSelectQuery(data);
+    } else if (this.isUpsertQuery()) {
+      //Use the same return value as that of MySQL & Postgres
+      if (data[0].$action === 'INSERT') {
+        result = 1;
+      } else {
+        result = 2;
+      }
     } else if (this.isCallQuery()) {
       result = data[0];
     } else if (this.isBulkUpdateQuery()) {

lib/model.js

@@ -2069,7 +2069,7 @@ class Model {
    * * MySQL - Implemented as a single query `INSERT values ON DUPLICATE KEY UPDATE values`
    * * PostgreSQL - Implemented as a temporary function with exception handling: INSERT EXCEPTION WHEN unique_constraint UPDATE
    * * SQLite - Implemented as two queries `INSERT; UPDATE`. This means that the update is executed regardless of whether the row already existed or not
-   *
+   * * MSSQL - Implemented as a single query using `MERGE` and `WHEN (NOT) MATCHED THEN`
    * **Note** that SQLite returns undefined for created, no matter if the row was created or updated. This is because SQLite always runs INSERT OR IGNORE + UPDATE, in a single query, so there is no way to know whether the row was inserted or not.
    *
    * __Alias__: _insertOrUpdate_

test/integration/hooks/upsert.test.js

@@ -13,7 +13,8 @@ if (Support.sequelize.dialect.supports.upserts) {
       this.User = this.sequelize.define('User', {
         username: {
           type: DataTypes.STRING,
-          allowNull: false
+          allowNull: false,
+          unique: true //Either Primary Key/Unique Keys should be passed to upsert
         },
         mood: {
           type: DataTypes.ENUM,

test/integration/model/upsert.test.js

@@ -331,6 +331,49 @@ describe(Support.getTestDialectTeaser('Model'), function() {
         });
       });
 
+      it('Works when two separate uniqueKeys are passed', function() {
+        var User = this.sequelize.define('User', {
+          username: {
+            type: Sequelize.STRING,
+            unique: true
+          },
+          email: {
+            type: Sequelize.STRING,
+            unique: true
+          },
+          city: {
+            type: Sequelize.STRING
+          }
+        });
+        var clock = sinon.useFakeTimers();
+        return User.sync({ force: true }).bind(this).then(function() {
+          return User.upsert({ username: 'user1', email: 'user1@domain.ext', city: 'City' })
+            .then(function(created) {
+              if (dialect === 'sqlite') {
+                expect(created).to.be.undefined;
+              } else {
+                expect(created).to.be.ok;              
+              }
+              clock.tick(1000);            
+              return User.upsert({ username: 'user1', email: 'user1@domain.ext', city: 'New City' });
+            }).then(function(created) {
+              if (dialect === 'sqlite') {
+                expect(created).to.be.undefined;
+              } else {
+                expect(created).not.to.be.ok;              
+              }
+              clock.tick(1000);
+              return User.findOne({ where: { username: 'user1', email: 'user1@domain.ext' }});              
+            })
+            .then(function(user) {
+              expect(user.createdAt).to.be.ok;
+              expect(user.city).to.equal('New City');
+              expect(user.updatedAt).to.be.afterTime(user.createdAt);
+            });
+        });
+      });
+      
+
     });
   }
 });