Merge pull request #2090 from ebean-orm/feature/2089

#2089 - Postgres - Use NO KEY with FOR UPDATE clauses with Postgres

ebean-api/src/main/java/io/ebean/config/PlatformConfig.java

@@ -17,6 +17,11 @@ public class PlatformConfig {
 
   private boolean allQuotedIdentifiers;
 
+  /**
+   * Set this to true for Postgres FOR UPDATE to include the primary key (not use NO KEY).
+   */
+  private boolean lockWithKey;
+
   private DbConstraintNaming constraintNaming;
 
   /**
@@ -77,6 +82,7 @@ public class PlatformConfig {
    * Construct based on given config - typically for DbMigration generation with many platforms.
    */
   public PlatformConfig(PlatformConfig platformConfig) {
+    this.lockWithKey = platformConfig.lockWithKey;
     this.databaseBooleanFalse = platformConfig.databaseBooleanFalse;
     this.databaseBooleanTrue = platformConfig.databaseBooleanTrue;
     this.databaseSequenceBatchSize = platformConfig.databaseSequenceBatchSize;
@@ -133,14 +139,26 @@ public class PlatformConfig {
     this.caseSensitiveCollation = caseSensitiveCollation;
   }
 
+  /**
+   * Return true if Postgres FOR UPDATE should include the primary key (or use NO KEY).
+   */
+  public boolean isLockWithKey() {
+    return lockWithKey;
+  }
+
+  /**
+   * Set to true such that Postgres FOR UPDATE should include the primary key (not use NO KEY option).
+   */
+  public void setLockWithKey(boolean lockWithKey) {
+    this.lockWithKey = lockWithKey;
+  }
+
   /**
    * Return a value used to represent TRUE in the database.
    * <p>
    * This is used for databases that do not support boolean natively.
-   * </p>
    * <p>
    * The value returned is either a Integer or a String (e.g. "1", or "T").
-   * </p>
    */
   public String getDatabaseBooleanTrue() {
     return databaseBooleanTrue;
@@ -150,10 +168,8 @@ public class PlatformConfig {
    * Set the value to represent TRUE in the database.
    * <p>
    * This is used for databases that do not support boolean natively.
-   * </p>
    * <p>
    * The value set is either a Integer or a String (e.g. "1", or "T").
-   * </p>
    */
   public void setDatabaseBooleanTrue(String databaseBooleanTrue) {
     this.databaseBooleanTrue = databaseBooleanTrue;
@@ -245,7 +261,6 @@ public class PlatformConfig {
 
   /**
    * Add a custom type mapping.
-   * <p>
    * <pre>{@code
    *
    *   // set the default mapping for BigDecimal.class/decimal
@@ -266,7 +281,6 @@ public class PlatformConfig {
 
   /**
    * Add a custom type mapping that applies to all platforms.
-   * <p>
    * <pre>{@code
    *
    *   // set the default mapping for BigDecimal/decimal
@@ -294,6 +308,7 @@ public class PlatformConfig {
   public void loadSettings(PropertiesWrapper p) {
 
     idType = p.getEnum(IdType.class, "idType", idType);
+    lockWithKey = p.getBoolean("lockWithKey", lockWithKey);
     databaseSequenceBatchSize = p.getInt("databaseSequenceBatchSize", databaseSequenceBatchSize);
     databaseBooleanTrue = p.get("databaseBooleanTrue", databaseBooleanTrue);
     databaseBooleanFalse = p.get("databaseBooleanFalse", databaseBooleanFalse);
@@ -334,7 +349,6 @@ public class PlatformConfig {
    */
   public enum DbUuid {
 
-
     /**
      * Store using native UUID in H2 and Postgres and otherwise fallback to VARCHAR(40).
      */

ebean-api/src/main/java/io/ebean/config/dbplatform/postgres/PostgresPlatform.java

@@ -4,6 +4,7 @@ import io.ebean.BackgroundExecutor;
 import io.ebean.Query;
 import io.ebean.annotation.PartitionMode;
 import io.ebean.annotation.Platform;
+import io.ebean.config.PlatformConfig;
 import io.ebean.config.dbplatform.DatabasePlatform;
 import io.ebean.config.dbplatform.DbPlatformType;
 import io.ebean.config.dbplatform.DbType;
@@ -25,6 +26,11 @@ import java.sql.Types;
  */
 public class PostgresPlatform extends DatabasePlatform {
 
+  // by default using NO KEY option with FOR UPDATE clauses
+  private String forUpdateSkipLocked = " for no key update skip locked";
+  private String forUpdateNowait = " for no key update nowait";
+  private String forUpdate = " for no key update";
+
   public PostgresPlatform() {
     super();
     this.platform = Platform.POSTGRES;
@@ -81,6 +87,16 @@ public class PostgresPlatform extends DatabasePlatform {
     dbTypeMap.put(DbType.LONGVARCHAR, dbTypeText);
   }
 
+  @Override
+  public void configure(PlatformConfig config) {
+    super.configure(config);
+    if (config.isLockWithKey()) {
+      this.forUpdateSkipLocked = " for update skip locked";
+      this.forUpdateNowait = " for update nowait";
+      this.forUpdate = " for update";
+    }
+  }
+
   @Override
   protected void addGeoTypes(int srid) {
     dbTypeMap.put(DbType.POINT, geoType("point", srid));
@@ -116,11 +132,11 @@ public class PostgresPlatform extends DatabasePlatform {
   protected String withForUpdate(String sql, Query.ForUpdate forUpdateMode) {
     switch (forUpdateMode) {
       case SKIPLOCKED:
-        return sql + " for update skip locked";
+        return sql + forUpdateSkipLocked;
       case NOWAIT:
-        return sql + " for update nowait";
+        return sql + forUpdateNowait;
       default:
-        return sql + " for update";
+        return sql + forUpdate;
     }
   }
 

ebean-core/src/test/java/io/ebean/config/ServerConfigTest.java

@@ -73,6 +73,7 @@ public class ServerConfigTest {
     props.setProperty("caseSensitiveCollation", "false");
     props.setProperty("loadModuleInfo", "true");
     props.setProperty("collectQueryPlanThresholdMicros", "10000");
+    props.setProperty("lockWithKey", "true");
 
     serverConfig.loadFromProperties(props);
 
@@ -84,6 +85,7 @@ public class ServerConfigTest {
 
     assertTrue(serverConfig.isIdGeneratorAutomatic());
     assertFalse(serverConfig.getPlatformConfig().isCaseSensitiveCollation());
+    assertTrue(serverConfig.getPlatformConfig().isLockWithKey());
 
     assertThat(serverConfig.getNamingConvention()).isInstanceOf(MatchingNamingConvention.class);
 

ebean-core/src/test/java/io/ebean/config/dbplatform/PostgresPlatformTest.java

@@ -1,5 +1,6 @@
 package io.ebean.config.dbplatform;
 
+import io.ebean.Query;
 import io.ebean.config.PlatformConfig;
 import io.ebean.config.dbplatform.postgres.PostgresPlatform;
 import org.junit.Test;
@@ -20,4 +21,33 @@ public class PostgresPlatformTest {
     assertThat(columnDefn).isEqualTo("uuid");
   }
 
+  @Test
+  public void default_forUpdate_expect_noKeyUsed() {
+
+    DatabasePlatform platform = new PostgresPlatform();
+
+    PlatformConfig config = new PlatformConfig();
+    platform.configure(config);
+
+    assertThat(config.isLockWithKey()).isFalse();
+    assertThat(platform.withForUpdate("X", Query.ForUpdate.SKIPLOCKED)).isEqualTo("X for no key update skip locked");
+    assertThat(platform.withForUpdate("X", Query.ForUpdate.NOWAIT)).isEqualTo("X for no key update nowait");
+    assertThat(platform.withForUpdate("X", Query.ForUpdate.BASE)).isEqualTo("X for no key update");
+  }
+
+  @Test
+  public void lockWithKey_forUpdate() {
+
+    DatabasePlatform platform = new PostgresPlatform();
+
+    PlatformConfig config = new PlatformConfig();
+    config.setLockWithKey(true);
+    platform.configure(config);
+
+    assertThat(config.isLockWithKey()).isTrue();
+    assertThat(platform.withForUpdate("X", Query.ForUpdate.SKIPLOCKED)).isEqualTo("X for update skip locked");
+    assertThat(platform.withForUpdate("X", Query.ForUpdate.NOWAIT)).isEqualTo("X for update nowait");
+    assertThat(platform.withForUpdate("X", Query.ForUpdate.BASE)).isEqualTo("X for update");
+  }
+
 }

ebean-core/src/test/java/org/tests/basic/TestQueryForUpdate.java

@@ -21,7 +21,6 @@ import static org.junit.Assert.assertTrue;
 
 public class TestQueryForUpdate extends BaseTestCase {
 
-
   @Test
   @ForPlatform({Platform.H2, Platform.ORACLE, Platform.POSTGRES, Platform.SQLSERVER, Platform.MYSQL, Platform.MARIADB})
   public void testForUpdate() {
@@ -35,6 +34,8 @@ public class TestQueryForUpdate extends BaseTestCase {
     query.findList();
     if (isSqlServer()) {
       assertThat(sqlOf(query)).contains("with (updlock)");
+    } else if (isPostgres()) {
+      assertThat(sqlOf(query)).contains("for no key update");
     } else {
       assertThat(sqlOf(query)).contains("for update");
     }
@@ -67,7 +68,11 @@ public class TestQueryForUpdate extends BaseTestCase {
       if (isH2() || isPostgres()) {
         assertSql(sql.get(0)).contains("from e_basic t0 where t0.id =");
         assertSql(sql.get(1)).contains("from e_basic t0 where t0.id =");
-        assertSql(sql.get(1)).contains("for update");
+        if (isPostgres()) {
+          assertSql(sql.get(1)).contains("for no key update");
+        } else {
+          assertSql(sql.get(1)).contains("for update");
+        }
       }
 
       transaction.end();
@@ -91,6 +96,8 @@ public class TestQueryForUpdate extends BaseTestCase {
       assertThat(sqlOf(query)).contains("for update");
     } else if (isSqlServer()) {
       assertThat(sqlOf(query)).contains("with (updlock,nowait)");
+    } else if (isPostgres()) {
+      assertThat(sqlOf(query)).contains("for no key update nowait");
     } else {
       assertThat(sqlOf(query)).contains("for update nowait");
     }
@@ -116,6 +123,8 @@ public class TestQueryForUpdate extends BaseTestCase {
         assertThat(sqlOf(query)).contains("with (updlock,nowait)");
       } else if (isH2()) {
         assertThat(sqlOf(query)).contains("for update");
+      } else if (isPostgres()) {
+        assertThat(sqlOf(query)).contains("for no key update nowait");
       } else {
         assertThat(sqlOf(query)).contains("for update nowait");
       }

ebean-core/src/test/java/org/tests/basic/TestQueryForUpdatePostgresLock.java

@@ -0,0 +1,86 @@
+package org.tests.basic;
+
+import io.ebean.BaseTestCase;
+import io.ebean.DB;
+import io.ebean.annotation.ForPlatform;
+import io.ebean.annotation.Platform;
+import io.ebean.annotation.Transactional;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.tests.model.basic.Article;
+import org.tests.model.basic.Section;
+
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+public class TestQueryForUpdatePostgresLock extends BaseTestCase {
+
+  private static final Logger log = LoggerFactory.getLogger(TestQueryForUpdatePostgresLock.class);
+
+  private long timePreInsert;
+  private long timePostInsert;
+  private long timePreLock;
+  private long timePostLock;
+
+  @Test
+  @ForPlatform(Platform.POSTGRES)
+  public void testForUpdatePostgresLock() throws InterruptedException {
+
+    Article article = new Article("lockTest", "auth");
+    DB.save(article);
+    final Integer id = article.getId();
+
+    ExecutorService exec = Executors.newFixedThreadPool(2);
+    exec.submit(() -> lockArticle(id));
+    exec.submit(() -> insertSection(id));
+    exec.awaitTermination(2, TimeUnit.SECONDS);
+    exec.shutdown();
+
+    // assert the lock was obtained before the insert was attempted
+    assertThat(timePreLock).isLessThan(timePreInsert);
+    // assert that the insert wasn't waiting on the lock to complete
+    assertThat(timePostInsert).isLessThan(timePostLock);
+  }
+
+  /**
+   * This holds row lock on article for 1 second.
+   * With FOR NO KEY UPDATE this does not block the insert.
+   */
+  @Transactional
+  private void lockArticle(Integer id) {
+    timePreLock = System.currentTimeMillis();
+    log.info("lock start");
+    DB.find(Article.class).setId(id).forUpdate().findOne();
+    sleep(1000);
+    timePostLock = System.currentTimeMillis();
+    log.info("lock done");
+  }
+
+  /**
+   * This inserts with FK to the article that is locked.
+   * With FOR NO KEY UPDATE this insert does not wait on the lock.
+   */
+  @Transactional
+  private void insertSection(Integer id) {
+    sleep(100);
+    log.info("insert start");
+    timePreInsert = System.currentTimeMillis();
+    Section section = new Section();
+    section.setArticle(DB.getReference(Article.class, id));
+    DB.save(section);
+    timePostInsert = System.currentTimeMillis();
+    log.info("inserted");
+  }
+
+  private void sleep(int millis) {
+    try {
+      Thread.sleep(millis);
+    } catch (InterruptedException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}