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

HTS concurrent put and delete resolution -- Reopen and Refractor tests

services/housetables/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;
 

services/housetables/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;
@@ -19,7 +20,16 @@ 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();

services/housetables/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());

services/housetables/src/test/java/com/linkedin/openhouse/housetables/e2e/usertable/HtsControllerTest.java

@@ -47,7 +47,9 @@ public class HtsControllerTest {
   public void setup() {
     // TODO: Use rest API to create the table and test the find/delete user table again.
     // For now manually create the user table upfront.
-    htsRepository.save(TestHouseTableModelConstants.TEST_USER_TABLE_ROW);
+    UserTableRow testUserTableRow =
+        new TestHouseTableModelConstants.TestTuple(0).get_userTableRow();
+    htsRepository.save(testUserTableRow);
   }
 
   @AfterEach

services/housetables/src/test/java/com/linkedin/openhouse/housetables/e2e/usertable/HtsRepositoryTest.java

@@ -9,10 +9,12 @@
 import com.linkedin.openhouse.housetables.model.UserTableRow;
 import com.linkedin.openhouse.housetables.model.UserTableRowPrimaryKey;
 import com.linkedin.openhouse.housetables.repository.HtsRepository;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Assertions;
 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,9 +24,26 @@ public class HtsRepositoryTest {
 
   @Autowired HtsRepository<UserTableRow, UserTableRowPrimaryKey> htsRepository;
 
+  @AfterEach
+  public void tearDown() {
+    htsRepository.deleteAll();
+  }
+
+  @Test
+  public void testSaveFirstRecord() {
+    UserTableRow testUserTableRow =
+        new TestHouseTableModelConstants.TestTuple(0).get_userTableRow();
+    // before insertion
+    Assertions.assertEquals(null, testUserTableRow.getVersion());
+    // after insertion
+    Assertions.assertEquals(0, htsRepository.save(testUserTableRow).getVersion());
+  }
+
   @Test
   public void testHouseTable() {
-    htsRepository.save(TEST_USER_TABLE_ROW);
+    UserTableRow testUserTableRow =
+        new TestHouseTableModelConstants.TestTuple(0).get_userTableRow();
+    htsRepository.save(testUserTableRow);
     UserTableRow actual =
         htsRepository
             .findById(
@@ -34,8 +53,7 @@ public void testHouseTable() {
                     .build())
             .orElse(UserTableRow.builder().build());
 
-    assertThat(isUserTableRowEqual(TestHouseTableModelConstants.TEST_USER_TABLE_ROW, actual))
-        .isTrue();
+    Assertions.assertEquals(testUserTableRow, actual);
     htsRepository.delete(actual);
   }
 
@@ -57,13 +75,21 @@ public void testDeleteUserTable() {
 
   @Test
   public void testSaveUserTableWithConflict() {
-    Long currentVersion = htsRepository.save(TEST_USER_TABLE_ROW).getVersion();
+    UserTableRow testUserTableRow =
+        new TestHouseTableModelConstants.TestTuple(0).get_userTableRow();
+    Long currentVersion = htsRepository.save(testUserTableRow).getVersion();
+    // test create the table again
+    Exception exception =
+        Assertions.assertThrows(
+            Exception.class,
+            () -> htsRepository.save(testUserTableRow.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()));
+            () -> htsRepository.save(testUserTableRow.toBuilder().version(100L).build()));
     Assertions.assertTrue(
         exception instanceof ObjectOptimisticLockingFailureException
             | exception instanceof EntityConcurrentModificationException);
@@ -72,7 +98,7 @@ public void testSaveUserTableWithConflict() {
     Assertions.assertNotEquals(
         htsRepository
             .save(
-                TEST_USER_TABLE_ROW
+                testUserTableRow
                     .toBuilder()
                     .version(currentVersion)
                     .metadataLocation("file:/ml2")
@@ -82,16 +108,12 @@ public void testSaveUserTableWithConflict() {
 
     // test update at older version
     exception =
-        Assertions.assertThrows(Exception.class, () -> htsRepository.save(TEST_USER_TABLE_ROW));
+        Assertions.assertThrows(Exception.class, () -> htsRepository.save(testUserTableRow));
     Assertions.assertTrue(
         exception instanceof ObjectOptimisticLockingFailureException
             | exception instanceof EntityConcurrentModificationException);
 
     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());
-  }
 }

services/housetables/src/test/java/com/linkedin/openhouse/housetables/e2e/usertable/UserTablesServiceTest.java

@@ -37,7 +37,9 @@ public class UserTablesServiceTest {
 
   @BeforeEach
   public void setup() {
-    htsRepository.save(TestHouseTableModelConstants.TEST_USER_TABLE_ROW);
+    UserTableRow testUserTableRow =
+        new TestHouseTableModelConstants.TestTuple(0).get_userTableRow();
+    htsRepository.save(testUserTableRow);
     htsRepository.save(testTuple1_0.get_userTableRow());
     htsRepository.save(testTuple2_0.get_userTableRow());
     htsRepository.save(testTuple1_1.get_userTableRow());

services/housetables/src/test/java/com/linkedin/openhouse/housetables/mock/mapper/UserTableVersionMapperTest.java

@@ -3,6 +3,7 @@
 import com.linkedin.openhouse.common.exception.EntityConcurrentModificationException;
 import com.linkedin.openhouse.housetables.dto.mapper.UserTableVersionMapper;
 import com.linkedin.openhouse.housetables.model.TestHouseTableModelConstants;
+import com.linkedin.openhouse.housetables.model.UserTableRow;
 import java.util.Optional;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -17,30 +18,42 @@ 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
   void testToVersionWithExistingRowAndCorrectMetadataLocation() {
+    UserTableRow testUserTableRow =
+        new TestHouseTableModelConstants.TestTuple(0).get_userTableRow();
     Assertions.assertEquals(
         versionMapper.toVersion(
             TestHouseTableModelConstants.TEST_USER_TABLE
                 .toBuilder()
-                .tableVersion(
-                    TestHouseTableModelConstants.TEST_USER_TABLE_ROW.getMetadataLocation())
+                .tableVersion(testUserTableRow.getMetadataLocation())
                 .build(),
-            Optional.of(TestHouseTableModelConstants.TEST_USER_TABLE_ROW)),
-        TestHouseTableModelConstants.TEST_USER_TABLE_ROW.getVersion());
+            Optional.of(testUserTableRow)),
+        testUserTableRow.getVersion());
   }
 
   @Test
   void testToVersionWithExistingRowAndIncorrectMetadataLocation() {
+    UserTableRow testUserTableRow =
+        new TestHouseTableModelConstants.TestTuple(0).get_userTableRow();
     Assertions.assertThrows(
         EntityConcurrentModificationException.class,
         () ->
             versionMapper.toVersion(
-                TestHouseTableModelConstants.TEST_USER_TABLE,
-                Optional.of(TestHouseTableModelConstants.TEST_USER_TABLE_ROW)));
+                TestHouseTableModelConstants.TEST_USER_TABLE, Optional.of(testUserTableRow)));
   }
 }

services/housetables/src/test/java/com/linkedin/openhouse/housetables/mock/mapper/UserTablesMapperTest.java

@@ -3,6 +3,7 @@
 import com.linkedin.openhouse.housetables.dto.mapper.UserTablesMapper;
 import com.linkedin.openhouse.housetables.dto.model.UserTableDto;
 import com.linkedin.openhouse.housetables.model.TestHouseTableModelConstants;
+import com.linkedin.openhouse.housetables.model.UserTableRow;
 import java.util.Optional;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
@@ -15,8 +16,9 @@ public class UserTablesMapperTest {
 
   @Test
   void toUserTableDto() {
-    UserTableDto dtoAfterMapping =
-        userTablesMapper.toUserTableDto(TestHouseTableModelConstants.TEST_USER_TABLE_ROW);
+    UserTableRow testUserTableRow =
+        new TestHouseTableModelConstants.TestTuple(0).get_userTableRow();
+    UserTableDto dtoAfterMapping = userTablesMapper.toUserTableDto(testUserTableRow);
     // Assert objects are equal ignoring versions
     Assertions.assertEquals(
         TestHouseTableModelConstants.TEST_USER_TABLE_DTO.toBuilder().tableVersion("").build(),
@@ -36,17 +38,21 @@ void toUserTable() {
 
   @Test
   void toUserTableRowNullStorageType() {
+    UserTableRow testUserTableRow =
+        new TestHouseTableModelConstants.TestTuple(0).get_userTableRow();
     Assertions.assertEquals(
-        TestHouseTableModelConstants.TEST_USER_TABLE_ROW,
+        testUserTableRow,
         userTablesMapper.toUserTableRow(
             TestHouseTableModelConstants.TEST_USER_TABLE.toBuilder().storageType(null).build(),
             Optional.empty()));
   }
 
   @Test
   void toUserTableRowCustomStorageType() {
+    UserTableRow testUserTableRow =
+        new TestHouseTableModelConstants.TestTuple(0).get_userTableRow();
     Assertions.assertEquals(
-        TestHouseTableModelConstants.TEST_USER_TABLE_ROW.toBuilder().storageType("blobfs").build(),
+        testUserTableRow.toBuilder().storageType("blobfs").build(),
         userTablesMapper.toUserTableRow(
             TestHouseTableModelConstants.TEST_USER_TABLE.toBuilder().storageType("blobfs").build(),
             Optional.empty()));

services/housetables/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;
@@ -19,7 +20,6 @@ private TestHouseTableModelConstants() {
 
   public static final String TEST_DEFAULT_STORAGE_TYPE = "hdfs";
 
-  public static final UserTableRow TEST_USER_TABLE_ROW = tuple0.get_userTableRow();
   public static final UserTableDto TEST_USER_TABLE_DTO = tuple0.get_userTableDto();
   public static final UserTable TEST_USER_TABLE = tuple0.get_userTable();
 
@@ -51,16 +51,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,7 +80,7 @@ public TestTuple(int tbSeq, int dbSeq) {
           UserTableRow.builder()
               .tableId(tableId)
               .databaseId(databaseId)
-              .version(1L)
+              .version(null)
               .metadataLocation(tableLoc)
               .storageType(storageType)
               .build();