Refactored the upsert query for postgres to be atomic

lib/dialects/postgres/query-generator.js

@@ -321,16 +321,50 @@ module.exports = (function() {
       });
       });
     },
     },
 
 
-    // http://www.maori.geek.nz/post/postgres_upsert_update_or_insert_in_ger_using_knex_js
-    upsertQuery: function (tableName, insertValues, updateValues, where, rawAttributes, options) {
-      var query = 'WITH upsert AS (<%= update %> RETURNING *) <%= insert %> WHERE NOT EXISTS (SELECT * FROM upsert)';
+    fn: function(fnName, tableName, body, returns, language) {
+      fnName = fnName || 'testfunc';
+      language = language || 'plpgsql';
+      returns = returns || 'SETOF ' + this.quoteTable(tableName);
+
+      var query = 'CREATE OR REPLACE FUNCTION pg_temp.<%= fnName %>() RETURNS <%= returns %> AS $$ BEGIN <%= body %> END; $$ LANGUAGE <%= language %>; SELECT * FROM pg_temp.<%= fnName %>();';
 
 
       return Utils._.template(query)({
       return Utils._.template(query)({
-        update: this.updateQuery(tableName, updateValues, where, options, rawAttributes),
-        insert: this.insertQuery(tableName, insertValues, rawAttributes, options).replace(/VALUES \((.*?)\);/, 'SELECT $1')
+        fnName: fnName,
+        returns: returns,
+        language: language,
+        body: body
       });
       });
     },
     },
 
 
+    exceptionFn: function(fnName, tableName, main, then, when, returns, language) {
+      when  = when || 'unique_violation';
+
+      var body = '<%= main %> EXCEPTION WHEN <%= when %> THEN <%= then %>;';
+      body = Utils._.template(body, {
+        main: main,
+        when: when,
+        then: then
+      });
+
+      return this.fn(fnName, tableName, body, returns, language);
+    },
+
+    // http://www.maori.geek.nz/post/postgres_upsert_update_or_insert_in_ger_using_knex_js
+    upsertQuery: function (tableName, insertValues, updateValues, where, rawAttributes, options) {
+      var insert = this.insertQuery(tableName, insertValues, rawAttributes, options).replace(/VALUES \((.*?)\)/, 'SELECT $1');
+      var update = this.updateQuery(tableName, updateValues, where, options, rawAttributes);
+
+      // The numbers here are selected to match the number of affected rows returned by MySQL
+      return this.exceptionFn(
+        'sequelize_upsert',
+        tableName,
+        insert + " RETURN 1;",
+        update + "; RETURN 2",
+        'unique_violation',
+        'integer'
+      );
+    },
+
     bulkInsertQuery: function(tableName, attrValueHashes, options, modelAttributes) {
     bulkInsertQuery: function(tableName, attrValueHashes, options, modelAttributes) {
       options = options || {};
       options = options || {};
 
 

lib/dialects/postgres/query.js

@@ -210,8 +210,10 @@ module.exports = (function() {
         }
         }
 
 
         return self.handleSelectQuery(rows);
         return self.handleSelectQuery(rows);
-      } else if ([QueryTypes.BULKDELETE, QueryTypes.UPSERT].indexOf(self.options.type) !== -1) {
+      } else if (QueryTypes.BULKDELETE === self.options.type) {
         return result.rowCount;
         return result.rowCount;
+      } else if (self.isUpsertQuery()) {
+        return rows[0].sequelize_upsert;
       } else if (self.isInsertQuery() || self.isUpdateQuery()) {
       } else if (self.isInsertQuery() || self.isUpdateQuery()) {
         if (!!self.callee && self.callee.dataValues) {
         if (!!self.callee && self.callee.dataValues) {
           if (!!self.callee.Model && !!self.callee.Model._hasHstoreAttributes) {
           if (!!self.callee.Model && !!self.callee.Model._hasHstoreAttributes) {

lib/model.js

@@ -1179,7 +1179,7 @@ module.exports = (function() {
    * **Implementation details:**
    * **Implementation details:**
    *
    *
    * * MySQL - Implemented as a single query `INSERT values ON DUPLICATE KEY UPDATE values`
    * * MySQL - Implemented as a single query `INSERT values ON DUPLICATE KEY UPDATE values`
-   * * PostgreSQL - Implemented as two queries `WITH upsert AS (update_query) insert_query WHERE NOT EXISTS (SELECT * FROM upsert)`, as outlined in // http://www.maori.geek.nz/post/postgres_upsert_update_or_insert_in_ger_using_knex_js
+   * * 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
    * * SQLite - Implemented as two queries `INSERT; UPDATE`. This means that the update is executed regardless of whether the row already existed or not
    *
    *
    * **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.
    * **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.
@@ -1188,6 +1188,7 @@ module.exports = (function() {
    * @param  {Object} [options]
    * @param  {Object} [options]
    * @param  {Array}  [options.fields=Object.keys(this.attributes)] The fields to insert / update. Defaults to all fields
    * @param  {Array}  [options.fields=Object.keys(this.attributes)] The fields to insert / update. Defaults to all fields
    *
    *
+   * @alias insertOrUpdate
    * @return {Promise<created>} Returns a boolean indicating whether the row was created or updated.
    * @return {Promise<created>} Returns a boolean indicating whether the row was created or updated.
    */
    */
   Model.prototype.upsert = function (values, options) {
   Model.prototype.upsert = function (values, options) {
@@ -1220,6 +1221,8 @@ module.exports = (function() {
     return this.QueryInterface.upsert(this.getTableName(), values, this, options);
     return this.QueryInterface.upsert(this.getTableName(), values, this, options);
   };
   };
 
 
+  Model.prototype.insertOrUpdate = Model.prototype.upsert;
+
   /**
   /**
    * Create and insert multiple instances in bulk.
    * Create and insert multiple instances in bulk.
    *
    *

lib/query-interface.js

@@ -503,7 +503,9 @@ module.exports = (function() {
         return rowCount;
         return rowCount;
       }
       }
 
 
-      return rowCount > 0;
+      // MySQL returns 1 for inserted, 2 for updated http://dev.mysql.com/doc/refman/5.0/en/insert-on-duplicate.html. Postgres has been modded to do the same
+
+      return rowCount === 1;
     });
     });
   };
   };