Merge pull request #75 from jiang95-dev/hts-concurrency

linkedin · Apr 24, 2024 · 02b5577 · 02b5577
2 parents da7eedd + 245d036
commit 02b5577
Show file tree

Hide file tree

Showing 7 changed files with 75 additions and 29 deletions.
diff --git a/...ousetables/src/main/java/com/linkedin/openhouse/housetables/api/spec/model/UserTable.java b/...ousetables/src/main/java/com/linkedin/openhouse/housetables/api/spec/model/UserTable.java
@@ -37,7 +37,9 @@ public class UserTable {
   @Pattern(regexp = ALPHA_NUM_UNDERSCORE_REGEX, message = ALPHA_NUM_UNDERSCORE_ERROR_MSG)
   private String databaseId;
 
-  @Schema(description = "Current Version of the user table.", example = "")
+  @Schema(
+      description = "Current Version of the user table. New record should have 'INTITAL_VERISON'",
+      example = "")
   @JsonProperty(value = "tableVersion")
   private String tableVersion;
 

diff --git a/...s/src/main/java/com/linkedin/openhouse/housetables/dto/mapper/UserTableVersionMapper.java b/...s/src/main/java/com/linkedin/openhouse/housetables/dto/mapper/UserTableVersionMapper.java
@@ -1,5 +1,6 @@
 package com.linkedin.openhouse.housetables.dto.mapper;
 
+import com.linkedin.openhouse.common.api.validator.ValidatorConstants;
 import com.linkedin.openhouse.common.exception.EntityConcurrentModificationException;
 import com.linkedin.openhouse.housetables.api.spec.model.UserTable;
 import com.linkedin.openhouse.housetables.model.UserTableRow;
@@ -15,11 +16,19 @@
  */
 @Mapper(componentModel = "spring")
 public class UserTableVersionMapper {
-
   @Named("toVersion")
   public Long toVersion(UserTable userTable, @Context Optional<UserTableRow> existingUserTableRow) {
     if (!existingUserTableRow.isPresent()) {
-      return 1L;
+      if (!userTable.getTableVersion().equals(ValidatorConstants.INITIAL_TABLE_VERSION)) {
+        throw new EntityConcurrentModificationException(
+            String.format(
+                "databaseId : %s, tableId : %s %s",
+                userTable.getDatabaseId(),
+                userTable.getTableId(),
+                "The requested user table has been deleted by other processes."),
+            new RuntimeException());
+      }
+      return null;
     } else {
       if (existingUserTableRow.get().getMetadataLocation().equals(userTable.getTableVersion())) {
         return existingUserTableRow.get().getVersion();

diff --git a/...bles/src/main/java/com/linkedin/openhouse/housetables/services/UserTablesServiceImpl.java b/...bles/src/main/java/com/linkedin/openhouse/housetables/services/UserTablesServiceImpl.java
@@ -15,6 +15,7 @@
 import java.util.stream.StreamSupport;
 import org.apache.iceberg.exceptions.CommitFailedException;
 import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.data.util.Pair;
 import org.springframework.orm.ObjectOptimisticLockingFailureException;
 import org.springframework.stereotype.Component;
@@ -68,7 +69,9 @@ public Pair<UserTableDto, Boolean> putUserTable(UserTable userTable) {
 
     try {
       returnedDto = userTablesMapper.toUserTableDto(htsJdbcRepository.save(targetUserTableRow));
-    } catch (CommitFailedException | ObjectOptimisticLockingFailureException ce) {
+    } catch (CommitFailedException
+        | ObjectOptimisticLockingFailureException
+        | DataIntegrityViolationException e) {
       throw new EntityConcurrentModificationException(
           String.format(
               "databaseId : %s, tableId : %s, version: %s %s",
@@ -77,7 +80,7 @@ public Pair<UserTableDto, Boolean> putUserTable(UserTable userTable) {
               targetUserTableRow.getVersion(),
               "The requested user table has been modified/created by other processes."),
           userTablesMapper.fromUserTableToRowKey(userTable).toString(),
-          ce);
+          e);
     }
 
     return Pair.of(returnedDto, existingUserTableRow.isPresent());

diff --git a/...les/src/test/java/com/linkedin/openhouse/housetables/e2e/usertable/HtsRepositoryTest.java b/...les/src/test/java/com/linkedin/openhouse/housetables/e2e/usertable/HtsRepositoryTest.java
@@ -13,6 +13,7 @@
 import org.junit.jupiter.api.Test;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.test.context.SpringBootTest;
+import org.springframework.dao.DataIntegrityViolationException;
 import org.springframework.orm.ObjectOptimisticLockingFailureException;
 import org.springframework.test.context.ContextConfiguration;
 
@@ -22,6 +23,14 @@ public class HtsRepositoryTest {
 
   @Autowired HtsRepository<UserTableRow, UserTableRowPrimaryKey> htsRepository;
 
+  @Test
+  public void testSaveFirstRecord() {
+    // before insertion
+    Assertions.assertEquals(null, TEST_USER_TABLE_ROW.getVersion());
+    // after insertion
+    Assertions.assertEquals(0, htsRepository.save(TEST_USER_TABLE_ROW).getVersion());
+  }
+
   @Test
   public void testHouseTable() {
     htsRepository.save(TEST_USER_TABLE_ROW);
@@ -58,9 +67,15 @@ public void testDeleteUserTable() {
   @Test
   public void testSaveUserTableWithConflict() {
     Long currentVersion = htsRepository.save(TEST_USER_TABLE_ROW).getVersion();
+    // test create the table again
+    Exception exception =
+        Assertions.assertThrows(
+            Exception.class,
+            () -> htsRepository.save(TEST_USER_TABLE_ROW.toBuilder().version(null).build()));
+    Assertions.assertTrue(exception instanceof DataIntegrityViolationException);
 
     // test update at wrong version
-    Exception exception =
+    exception =
         Assertions.assertThrows(
             Exception.class,
             () -> htsRepository.save(TEST_USER_TABLE_ROW.toBuilder().version(100L).build()));
@@ -90,8 +105,4 @@ public void testSaveUserTableWithConflict() {
     htsRepository.deleteById(
         UserTableRowPrimaryKey.builder().databaseId(TEST_DB_ID).tableId(TEST_TABLE_ID).build());
   }
-
-  private Boolean isUserTableRowEqual(UserTableRow expected, UserTableRow actual) {
-    return expected.toBuilder().version(0L).build().equals(actual.toBuilder().version(0L).build());
-  }
 }
diff --git a/.../test/java/com/linkedin/openhouse/housetables/mock/mapper/UserTableVersionMapperTest.java b/.../test/java/com/linkedin/openhouse/housetables/mock/mapper/UserTableVersionMapperTest.java
@@ -17,8 +17,18 @@ public class UserTableVersionMapperTest {
   @Test
   void testToVersionWithNoExistingRow() {
     Assertions.assertEquals(
-        versionMapper.toVersion(TestHouseTableModelConstants.TEST_USER_TABLE, Optional.empty()),
-        1L);
+        null,
+        versionMapper.toVersion(TestHouseTableModelConstants.TEST_USER_TABLE, Optional.empty()));
+  }
+
+  @Test
+  void testToVersionWithNoExistingRowAndIncorrectTableVersion() {
+    Assertions.assertThrows(
+        EntityConcurrentModificationException.class,
+        () ->
+            versionMapper.toVersion(
+                TestHouseTableModelConstants.TEST_USER_TABLE.toBuilder().tableVersion("v1").build(),
+                Optional.empty()));
   }
 
   @Test

diff --git a/...es/src/test/java/com/linkedin/openhouse/housetables/mock/mapper/UserTablesMapperTest.java b/...es/src/test/java/com/linkedin/openhouse/housetables/mock/mapper/UserTablesMapperTest.java
@@ -1,5 +1,7 @@
 package com.linkedin.openhouse.housetables.mock.mapper;
 
+import static com.linkedin.openhouse.housetables.model.TestHouseTableModelConstants.*;
+
 import com.linkedin.openhouse.housetables.dto.mapper.UserTablesMapper;
 import com.linkedin.openhouse.housetables.dto.model.UserTableDto;
 import com.linkedin.openhouse.housetables.model.TestHouseTableModelConstants;
@@ -36,20 +38,28 @@ void toUserTable() {
 
   @Test
   void toUserTableRowNullStorageType() {
-    Assertions.assertEquals(
-        TestHouseTableModelConstants.TEST_USER_TABLE_ROW,
-        userTablesMapper.toUserTableRow(
-            TestHouseTableModelConstants.TEST_USER_TABLE.toBuilder().storageType(null).build(),
-            Optional.empty()));
+    Assertions.assertTrue(
+        isUserTableRowEqual(
+            TestHouseTableModelConstants.TEST_USER_TABLE_ROW,
+            userTablesMapper.toUserTableRow(
+                TestHouseTableModelConstants.TEST_USER_TABLE.toBuilder().storageType(null).build(),
+                Optional.empty())));
   }
 
   @Test
   void toUserTableRowCustomStorageType() {
-    Assertions.assertEquals(
-        TestHouseTableModelConstants.TEST_USER_TABLE_ROW.toBuilder().storageType("blobfs").build(),
-        userTablesMapper.toUserTableRow(
-            TestHouseTableModelConstants.TEST_USER_TABLE.toBuilder().storageType("blobfs").build(),
-            Optional.empty()));
+    Assertions.assertTrue(
+        isUserTableRowEqual(
+            TestHouseTableModelConstants.TEST_USER_TABLE_ROW
+                .toBuilder()
+                .storageType("blobfs")
+                .build(),
+            userTablesMapper.toUserTableRow(
+                TestHouseTableModelConstants.TEST_USER_TABLE
+                    .toBuilder()
+                    .storageType("blobfs")
+                    .build(),
+                Optional.empty())));
   }
 
   @Test

diff --git a/.../src/test/java/com/linkedin/openhouse/housetables/model/TestHouseTableModelConstants.java b/.../src/test/java/com/linkedin/openhouse/housetables/model/TestHouseTableModelConstants.java
@@ -1,5 +1,6 @@
 package com.linkedin.openhouse.housetables.model;
 
+import com.linkedin.openhouse.common.api.validator.ValidatorConstants;
 import com.linkedin.openhouse.housetables.api.spec.model.UserTable;
 import com.linkedin.openhouse.housetables.dto.model.UserTableDto;
 import lombok.Getter;
@@ -51,16 +52,12 @@ public TestTuple(int tbSeq) {
     public TestTuple(int tbSeq, int dbSeq) {
       this.tableId = "test_table" + tbSeq;
       this.databaseId = "test_db" + dbSeq;
-      this.ver =
-          LOC_TEMPLATE
-              .replace("$test_db", databaseId)
-              .replace("$test_table", tableId)
-              .replace("$version", "v0");
+      this.ver = ValidatorConstants.INITIAL_TABLE_VERSION;
       this.tableLoc =
           LOC_TEMPLATE
               .replace("$test_db", databaseId)
               .replace("$test_table", tableId)
-              .replace("$version", "v1");
+              .replace("$version", "v0");
       this.storageType = TEST_DEFAULT_STORAGE_TYPE;
       this._userTable =
           UserTable.builder()
@@ -84,10 +81,14 @@ public TestTuple(int tbSeq, int dbSeq) {
           UserTableRow.builder()
               .tableId(tableId)
               .databaseId(databaseId)
-              .version(1L)
+              .version(null)
               .metadataLocation(tableLoc)
               .storageType(storageType)
               .build();
     }
   }
+
+  public static Boolean isUserTableRowEqual(UserTableRow expected, UserTableRow actual) {
+    return expected.toBuilder().version(0L).build().equals(actual.toBuilder().version(0L).build());
+  }
 }