WIP: Implement "skipLocked()" and "noWait()"

src/dialects/mysql/query/compiler.js

@@ -39,6 +39,16 @@ assign(QueryCompiler_MySQL.prototype, {
     return 'lock in share mode';
   },
 
+  // Only supported on MySQL 8.0+
+  skipLocked() {
+    return 'skip locked';
+  },
+
+  // Supported on MySQL 8.0+ and MariaDB 10.3.0+
+  noWait() {
+    return 'nowait';
+  },
+
   // Compiles a `columnInfo` query.
   columnInfo() {
     const column = this.single.columnInfo;

src/dialects/postgres/query/compiler.js

@@ -102,6 +102,14 @@ assign(QueryCompiler_PG.prototype, {
     );
   },
 
+  skipLocked() {
+    return 'skip locked';
+  },
+
+  noWait() {
+    return 'nowait';
+  },
+
   // Compiles a columnInfo query
   columnInfo() {
     const column = this.single.columnInfo;

src/query/builder.js

@@ -1061,6 +1061,42 @@ assign(Builder.prototype, {
     return this;
   },
 
+  // Skips locked rows when using a lock constraint.
+  skipLocked() {
+    const { _method } = this;
+    if (!includes(['pluck', 'first', 'select'], _method)) {
+      throw new Error(`Cannot chain .skipLocked() on "${_method}" query!`);
+    }
+    if (!includes(['forShare', 'forUpdate'], this._single.lock)) {
+      throw new Error(
+        '.skipLocked() can only be used after a call to .forShare() or .forUpdate()!'
+      );
+    }
+    if (this._single.waitMode === 'noWait') {
+      throw new Error('.skipLocked() cannot be used with .noWait()!');
+    }
+    this._single.waitMode = 'skipLocked';
+    return this;
+  },
+
+  // Causes error when acessing a locked row instead of waiting for it to be released.
+  noWait() {
+    const { _method } = this;
+    if (!includes(['pluck', 'first', 'select'], _method)) {
+      throw new Error(`Cannot chain .noWait() on "${_method}" query!`);
+    }
+    if (!includes(['forShare', 'forUpdate'], this._single.lock)) {
+      throw new Error(
+        '.noWait() can only be used after a call to .forShare() or .forUpdate()!'
+      );
+    }
+    if (this._single.waitMode === 'skipLocked') {
+      throw new Error('.noWait() cannot be used with .skipLocked()!');
+    }
+    this._single.waitMode = 'noWait';
+    return this;
+  },
+
   // Takes a JS object of methods to call and calls them
   fromJS(obj) {
     each(obj, (val, key) => {

src/query/compiler.js

@@ -48,6 +48,7 @@ const components = [
   'limit',
   'offset',
   'lock',
+  'waitMode',
 ];
 
 assign(QueryCompiler.prototype, {
@@ -542,6 +543,13 @@ assign(QueryCompiler.prototype, {
     }
   },
 
+  // Compiles the wait mode on the locks.
+  waitMode() {
+    if (this.single.waitMode) {
+      return this[this.single.waitMode]();
+    }
+  },
+
   // On Clause
   // ------
 

test/integration/builder/selects.js

@@ -1225,5 +1225,103 @@ module.exports = function(knex) {
             });
         });
     });
+
+    it('forUpdate().skipLocked() should return the first non-locked row', function() {
+      // TODO: enable this test on MySQL when updating test suite to MySQL 8.0
+      if (knex.client.driverName !== 'pg') {
+        return;
+      }
+
+      const rowName = 'row for skipLocked() test';
+      return knex('test_default_table')
+        .insert([
+          { string: rowName, tinyint: 1 },
+          { string: rowName, tinyint: 2 },
+        ])
+        .then(() => {
+          return knex
+            .transaction((trx) => {
+              // select and lock only the first row from this test
+              return trx('test_default_table')
+                .where({ string: rowName })
+                .orderBy('tinyint', 'asc')
+                .first()
+                .forUpdate()
+                .then((res) => {
+                  // try to lock the second row
+                  return knex('test_default_table')
+                    .where({ string: rowName })
+                    .orderBy('tinyint', 'asc')
+                    .forUpdate()
+                    .skipLocked()
+                    .first();
+                });
+            })
+            .then((res) => {
+              expect(res.tinyint).to.equal(2);
+            });
+        });
+    });
+
+    it('forUpdate().noWait() should reject immediately when a row is locked', function() {
+      // TODO: enable this test on MySQL when updating test suite to MySQL 8.0
+      if (knex.client.driverName !== 'pg') {
+        return;
+      }
+
+      const rowName = 'row for noWait() test';
+      return knex('test_default_table')
+        .insert([
+          { string: rowName, tinyint: 1 },
+          { string: rowName, tinyint: 2 },
+        ])
+        .then(() => {
+          return knex
+            .transaction((trx) => {
+              // select and lock only the first row from this test
+              return trx('test_default_table')
+                .where({ string: rowName })
+                .orderBy('tinyint', 'asc')
+                .first()
+                .forUpdate()
+                .then((res) => {
+                  // try to lock the second row
+                  return knex('test_default_table')
+                    .where({ string: rowName })
+                    .orderBy('tinyint', 'asc')
+                    .forUpdate()
+                    .noWait()
+                    .first();
+                });
+            })
+            .then((res) => {
+              expect(
+                'The query should have been cancelled when trying to select a locked row'
+              ).to.be.false;
+            })
+            .catch((err) => {
+              switch (knex.client.driverName) {
+                case 'pg':
+                  expect(err.message).to.contain(
+                    'could not obtain lock on row'
+                  );
+                  break;
+                case 'mysql':
+                case 'mysql2':
+                  // mysql
+                  expect(err.message).to.contain(
+                    'lock(s) could not be acquired immediately'
+                  );
+                  // mariadb
+                  // TODO: detect if test is being run on mysql or mariadb to check for the correct error message
+                  // expect(err.message).to.contain('Lock wait timeout exceeded');
+                  break;
+                default:
+                  // unsupported database
+                  throw err;
+              }
+            });
+        });
+    });
   });
 };

test/unit/query/builder.js

@@ -6377,6 +6377,40 @@ describe('QueryBuilder', function() {
     );
   });
 
+  it('lock for update with skip locked #1937', function() {
+    testsql(
+      qb()
+        .select('*')
+        .from('foo')
+        .first()
+        .forUpdate()
+        .skipLocked(),
+      {
+        pg: {
+          sql: 'select * from "foo" limit ? for update skip locked',
+          bindings: [1],
+        },
+      }
+    );
+  });
+
+  it('lock for update with nowait #1937', function() {
+    testsql(
+      qb()
+        .select('*')
+        .from('foo')
+        .first()
+        .forUpdate()
+        .noWait(),
+      {
+        pg: {
+          sql: 'select * from "foo" limit ? for update nowait',
+          bindings: [1],
+        },
+      }
+    );
+  });
+
   it('allows insert values of sub-select, #121', function() {
     testsql(
       qb()

types/knex.d.ts

@@ -464,6 +464,10 @@ declare namespace Knex {
     forShare(...tableNames: string[]): QueryBuilder;
     forShare(tableNames: string[]): QueryBuilder;
 
+    skipLocked(): QueryBuilder;
+
+    noWait(): QueryBuilder;
+
     toSQL(): Sql;
 
     on(event: string, callback: Function): QueryBuilder;