MSSQL upsert implementation - V3 backport (#6875)

.gitignore

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

changelog.md

+# Future
+- [ADDED] UPSERT support for MSSQL
+
 # 3.25.0
 - [FIXED] Set `timestamps` and `paranoid` options from through model on `belongsToMany` association
 - [FIXED] Properly apply paranoid condition when `groupedLimit.on` association is `paranoid`

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

@@ -301,6 +301,140 @@ var QueryGenerator = {
     return generatedQuery;
   },
 
+  upsertQuery: function(tableName, insertValues, updateValues, where, rawAttributes, options) {
+    var 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.*;';
+
+    var targetTableAlias = this.quoteTable(tableName + '_target')
+      , sourceTableAlias = this.quoteTable(tableName + '_source')
+      , primaryKeysAttrs = []
+      , identityAttrs = []
+      , uniqueAttrs = []
+      , tableNameQuoted = this.quoteTable(tableName)
+      , joinCondition
+      , needIdentityInsertWrapper = false;
+
+    
+    //Obtain primaryKeys, uniquekeys and identity attrs from rawAttributes as model is not passed
+    for (var 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);
+      }
+    }
+
+    var updateKeys = Object.keys(updateValues)
+      , insertKeys = Object.keys(insertValues);
+    
+    var insertKeysQuoted = Utils._.map(insertKeys, function(key) {
+      return this.quoteIdentifier(key);
+    }.bind(this)).join(', ');
+
+    var insertValuesEscaped = Utils._.map(insertKeys, function(key) {
+      return this.escape(insertValues[key]);
+    }.bind(this)).join(', ');
+    
+    var sourceTableQuery = 'VALUES(' + insertValuesEscaped + ')'; //Virtual Table
+
+    //IDENTITY_INSERT Condition
+    identityAttrs.forEach(function(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
+    var clauses = where.$or.filter(function(clause) {
+      var valid = true;
+      /*
+       * Exclude NULL Composite PK/UK. Partial Composite clauses should also be excluded as it doesn't guarantee a single row
+       */
+      for (var 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
+     */
+    var getJoinSnippet = function(array) {
+      return Utils._.map(array, function(key) {
+        key = this.quoteIdentifier(key);
+        return targetTableAlias + '.' + key + ' = ' + sourceTableAlias + '.' + key;
+      }.bind(this));
+    };
+
+    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 (key in clauses) {
+        var keys = Object.keys(clauses[key]);
+        if (primaryKeysAttrs.indexOf(keys[0]) !== -1) {
+          joinCondition = getJoinSnippet.bind(this)(primaryKeysAttrs).join(' AND ');
+          break;
+        }
+      }
+      if (!joinCondition) {
+        joinCondition = getJoinSnippet.bind(this)(uniqueAttrs).join(' AND ');
+      }
+    }
+    
+    // Remove the IDENTITY_INSERT Column from update
+    var updateSnippet = updateKeys.filter(function(key) {
+      if (identityAttrs.indexOf(key) === -1) {
+        return true;
+      } else {
+        return false;
+      }
+    });
+
+    updateSnippet = Utils._.map(updateSnippet, function(key) {
+      var value = this.escape(updateValues[key]);
+      key = this.quoteIdentifier(key);
+      return targetTableAlias + '.' + key + ' = ' + value;
+    }.bind(this)).join(', ');
+
+    var insertSnippet = '(' + insertKeysQuoted + ') VALUES(' + insertValuesEscaped + ')';
+    
+    var replacements = {
+      tableNameQuoted: tableNameQuoted,
+      targetTableAlias: targetTableAlias,
+      sourceTableQuery: sourceTableQuery,
+      sourceTableAlias: sourceTableAlias,
+      insertKeysQuoted: insertKeysQuoted,
+      joinCondition: joinCondition,
+      updateSnippet: updateSnippet,
+      insertSnippet: insertSnippet
+    };
+
+    query = Utils._.template(query)(replacements);
+
+
+    if (needIdentityInsertWrapper) {
+      query = [
+        'SET IDENTITY_INSERT', this.quoteTable(tableName), 'ON;',
+        query,
+        'SET IDENTITY_INSERT', this.quoteTable(tableName), 'OFF;',
+      ].join(' ');
+    }
+    return query;
+  },
+
   deleteQuery: function(tableName, where, options) {
     options = options || {};
 

lib/dialects/mssql/query.js

@@ -171,6 +171,13 @@ Query.prototype.formatResults = function(data) {
     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

@@ -1999,7 +1999,7 @@ Model.prototype.findCreateFind = function(options) {
  * * 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.
  *
  * @param  {Object}       values

test/config/config.js

@@ -27,7 +27,7 @@ module.exports = {
     username: process.env.SEQ_MSSQL_USER || process.env.SEQ_USER || 'sequelize',
     password: process.env.SEQ_MSSQL_PW   || process.env.SEQ_PW   || null,
     host:     process.env.SEQ_MSSQL_HOST || process.env.SEQ_HOST || '127.0.0.1',
-    port:     process.env.SEQ_MSSQL_PORT || process.env.SEQ_PORT || 11433,
+    port:     process.env.SEQ_MSSQL_PORT || process.env.SEQ_PORT || 1433,
     pool:     {
       maxConnections: process.env.SEQ_MSSQL_POOL_MAX  || process.env.SEQ_POOL_MAX  || 5,
       maxIdleTime:    process.env.SEQ_MSSQL_POOL_IDLE || process.env.SEQ_POOL_IDLE || 3000

test/integration/model/upsert.test.js

@@ -331,6 +331,47 @@ 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);
+            });
+        });
+      });
     });
   }
 });