improve rename logic to use 2 phase commit idea

metacat-common-server/src/main/java/com/netflix/metacat/common/server/usermetadata/ParentChildRelMetadataService.java

@@ -5,6 +5,8 @@
 import com.netflix.metacat.common.server.model.ChildInfo;
 import com.netflix.metacat.common.server.model.ParentInfo;
 import com.netflix.metacat.common.server.properties.ParentChildRelationshipProperties;
+import org.apache.commons.lang3.tuple.Pair;
+import java.util.Optional;
 
 import java.util.Set;
 
@@ -59,13 +61,16 @@ void deleteParentChildRelation(
     /**
      * Renames `oldName` to `newName` in the parentChildRelationship store.
      * This involves two steps:
-     * 1. Rename all records where the child is `oldName` to `newName`
-     * 2. Rename all records where the parent is `oldName` to `newName`
+     * 1. duplicate all records where the child is `oldName` with childName = `newName`
+     * 2. duplicate all records where the parent is `oldName` with parentName `newName`
      *
      * @param oldName  the current name to be renamed
      * @param newName  the new name to rename to
+     * @return return a pair of set,
+     * where the first set represents the affected parent_uuid with parent = oldName
+     * and the second set represents the affected child_uuid with child = oldName
      */
-    void rename(
+    Pair<Set<String>, Set<String>> renameSoftInsert(
         QualifiedName oldName,
         QualifiedName newName
     );
@@ -75,10 +80,13 @@ void rename(
      * This involves two steps:
      * 1. drop all records where the child column = `name`
      * 2. drop all records where the parent column = `name`
+     * * Note if uuids are specified, it is going to drop the table with that name with the corresponding uuids
      * @param name  the name of the entity to drop
+     * @param uuids the uuids to drop, where the first pair is the parent uuid and second pair is the child uuid
      */
     void drop(
-        QualifiedName name
+        QualifiedName name,
+        Optional<Pair<Set<String>, Set<String>>> uuids
     );
 
     /**

metacat-functional-tests/src/functionalTest/groovy/com/netflix/metacat/ParentChildRelMetadataServiceSpec.groovy

@@ -229,9 +229,13 @@ class ParentChildRelMetadataServiceSpec extends Specification{
         service.createParentChildRelation(parent, parentUUID, child, childUUID, type, props)
 
         when:
-        service.rename(parent, newParent)
+        def pair = service.renameSoftInsert(parent, newParent)
+        service.drop(parent, Optional.of(pair))
 
         then:
+        assert pair.getLeft().size() == 1
+        assert pair.getRight().isEmpty()
+
         // Test Old Parent Name
         assert service.getParents(parent).isEmpty()
         assert service.getChildren(parent).isEmpty()
@@ -254,10 +258,13 @@ class ParentChildRelMetadataServiceSpec extends Specification{
 
         // rename back
         when:
-        service.rename(newParent, parent)
+        pair = service.renameSoftInsert(newParent, parent)
+        service.drop(newParent, Optional.of(pair))
         child_parent_expected = [new ParentInfo(parent.toString(), type, parentUUID)] as Set
 
         then:
+        assert pair.getLeft().size() == 1
+        assert pair.getRight().isEmpty()
         // Test new Parent Name
         assert service.getParents(newParent).isEmpty()
         assert service.getChildren(newParent).isEmpty()
@@ -290,28 +297,29 @@ class ParentChildRelMetadataServiceSpec extends Specification{
         service.createParentChildRelation(parent, parentUUID, child2, child2UUID, type, props)
         def newParent = QualifiedName.ofTable(catalog, database, "np")
         def child_parent_expected = [new ParentInfo(newParent.toString(), type, parentUUID)] as Set
+        def newParent_children_expected = [
+            new ChildInfo(child1.toString(), type, child1UUID),
+            new ChildInfo(child2.toString(), type, child2UUID)
+        ] as Set
 
         when:
-        service.rename(parent, newParent)
+        def pair = service.renameSoftInsert(parent, newParent)
+        service.drop(parent, Optional.of(pair))
 
         then:
         // Test Old Parent Name
+        assert pair.getLeft().size() == 1
+        assert pair.getRight().isEmpty()
         assert service.getParents(parent).isEmpty()
         assert service.getChildren(parent).isEmpty()
         assert !service.isChildTable(parent)
         assert !service.isParentTable(parent)
 
         // Test New Parent Name
         assert service.getParents(newParent).isEmpty()
-        def newParent_children_expected = [
-            new ChildInfo(child1.toString(), type, child1UUID),
-            new ChildInfo(child2.toString(), type, child2UUID),
-        ] as Set
         assert service.getChildren(newParent) == newParent_children_expected
         assert !service.isChildTable(newParent)
         assert service.isParentTable(newParent)
-
-        then:
         // Test Child1
         assert service.getParents(child1) == child_parent_expected
         assert service.isChildTable(child1)
@@ -333,8 +341,12 @@ class ParentChildRelMetadataServiceSpec extends Specification{
         def newChild = QualifiedName.ofTable(catalog, database, "nc")
 
         when:
-        service.rename(child, newChild)
+        def pair = service.renameSoftInsert(child, newChild)
+        service.drop(child, Optional.of(pair))
+
         then:
+        assert pair.getLeft().isEmpty()
+        assert pair.getRight().size() == 1
         // Test Parent
         assert service.getParents(parent).isEmpty()
         def parent_children_expected  = [new ChildInfo(newChild.toString(), type, childUUID)] as Set
@@ -357,10 +369,13 @@ class ParentChildRelMetadataServiceSpec extends Specification{
 
         // rename back
         when:
-        service.rename(newChild, child)
+        pair = service.renameSoftInsert(newChild, child)
+        service.drop(newChild, Optional.of(pair))
         parent_children_expected  = [new ChildInfo(child.toString(), type, childUUID)] as Set
 
         then:
+        assert pair.getLeft().isEmpty()
+        assert pair.getRight().size() == 1
         // Test Parent
         assert service.getParents(parent).isEmpty()
         assert parent_children_expected == service.getChildren(parent)
@@ -390,7 +405,7 @@ class ParentChildRelMetadataServiceSpec extends Specification{
         def type = "clone";
         service.createParentChildRelation(parent, parentUUID, child, childUUID, type, props)
         when:
-        service.drop(child)
+        service.drop(child, Optional.empty())
 
         then:
         // Test Parent
@@ -417,8 +432,9 @@ class ParentChildRelMetadataServiceSpec extends Specification{
         service.createParentChildRelation(parent, parentUUID, child, childUUID, type, props)
 
         when:
-        service.rename(child, newChild)
-        service.drop(newChild)
+        def pair = service.renameSoftInsert(child, newChild)
+        service.drop(child, Optional.of(pair))
+        service.drop(newChild, Optional.empty())
 
         then:
         // Test Parent
@@ -460,14 +476,14 @@ class ParentChildRelMetadataServiceSpec extends Specification{
 
         // rename to an existing parent
         when:
-        service.rename(parent1, parent2)
+        service.renameSoftInsert(parent1, parent2)
         then:
         def e = thrown(Exception)
         assert e.message.contains("is already a parent table")
 
         // rename to an existing child
         when:
-        service.rename(child2, child1)
+        service.renameSoftInsert(child2, child1)
         then:
         e = thrown(Exception)
         assert e.message.contains("is already a child table")
@@ -618,4 +634,165 @@ class ParentChildRelMetadataServiceSpec extends Specification{
         1        |  "CLONE,5"               | "CLONE,test,3;OTHER,other,2"|  "CLONE,testhive/test/parent,2"                               | 2
         1        |  "CLONE,5;Other,3"       | "CLONE,test,3;CLONE,other,2"|  "CLONE,testhive/test/parent,2;CLONE,testhive/test/other,2"   | 2
     }
+    def "Test Parent Rename failed and remove newName"() {
+        setup:
+        def parent = QualifiedName.ofTable(catalog, database, "p")
+        def parentUUID = "p_uuid"
+        def child = QualifiedName.ofTable(catalog, database, "c")
+        def childUUID = "c_uuid"
+        def type = "clone";
+        def newParent = QualifiedName.ofTable(catalog, database, "np")
+        service.createParentChildRelation(parent, parentUUID, child, childUUID, type, props)
+
+        when:
+        def pair = service.renameSoftInsert(parent, newParent)
+        service.drop(newParent, Optional.of(pair))
+
+        then:
+        // Test New Parent Name
+        assert service.getParents(newParent).isEmpty()
+        assert service.getChildren(newParent).isEmpty()
+
+        // Test Old Parent Name
+        assert service.getParents(parent).isEmpty()
+        def newParent_children_expected = [new ChildInfo(child.toString(), type, childUUID)] as Set
+        assert service.getChildren(parent) == newParent_children_expected
+
+        // Test Child
+        def child_parent_expected = [new ParentInfo(parent.toString(), type, parentUUID)] as Set
+        assert child_parent_expected == service.getParents(child)
+        assert service.getChildren(child).isEmpty()
+    }
+
+    def "Test Child Rename failed and remove newName"() {
+        setup:
+        def parent = QualifiedName.ofTable(catalog, database, "p")
+        def parentUUID = "p_uuid"
+        def child = QualifiedName.ofTable(catalog, database, "c")
+        def childUUID = "c_uuid"
+        def type = "clone"
+        service.createParentChildRelation(parent, parentUUID, child, childUUID, type, props)
+        def newChild = QualifiedName.ofTable(catalog, database, "nc")
+
+        when:
+        def pair = service.renameSoftInsert(child, newChild)
+        service.drop(newChild, Optional.of(pair))
+        then:
+        // Test Parent
+        assert service.getParents(parent).isEmpty()
+        def parent_children_expected  = [new ChildInfo(child.toString(), type, childUUID)] as Set
+        assert parent_children_expected == service.getChildren(parent)
+
+        // Test New Child
+        assert service.getParents(newChild).isEmpty()
+        assert service.getChildren(newChild).isEmpty()
+
+        // Test Child
+        def child_parent_expected = [new ParentInfo(parent.toString(), type, parentUUID)] as Set
+        assert child_parent_expected == service.getParents(child)
+        assert service.getChildren(child).isEmpty()
+    }
+
+    def "Test rename to an existing table in the parent child rel service"() {
+        given:
+        def parent = QualifiedName.ofTable(catalog, database, "p")
+        def parentUUID = "p_uuid"
+        def child = QualifiedName.ofTable(catalog, database, "c")
+        def childUUID = "c_uuid"
+        def type = "clone"
+        service.createParentChildRelation(parent, parentUUID, child, childUUID, type, props)
+        def newParentQualifiedName = QualifiedName.ofTable(catalog, database, "np_name")
+        def newChildQualifiedName = QualifiedName.ofTable(catalog, database, "nc_name")
+        insertNewParentChildRecord(newParentQualifiedName.toString(), "np_uuid", newChildQualifiedName.toString(), "nc_uuid", "random")
+
+        when:
+        service.renameSoftInsert(parent, newParentQualifiedName)
+        then:
+        def e = thrown(RuntimeException)
+        assert e.message.contains("is already a parent table")
+
+        when:
+        service.renameSoftInsert(child, newChildQualifiedName)
+
+        then:
+        e = thrown(RuntimeException)
+        assert e.message.contains("is already a child table")
+    }
+
+    // Add a test where between rename and drop old/new name with the same name but different uuid is added
+    // only those get from rename should be drop
+    // This case is not possible in reality but wanted to add a test to prove this
+    def "Simulate Rename child: drop only impacted uuids"() {
+        setup:
+        def parent = QualifiedName.ofTable(catalog, database, "p")
+        def parentUUID = "p_uuid"
+        def child = QualifiedName.ofTable(catalog, database, "c")
+        def childUUID = "c_uuid"
+        def type = "clone"
+        service.createParentChildRelation(parent, parentUUID, child, childUUID, type, props)
+        def newChild = QualifiedName.ofTable(catalog, database, "nc")
+
+        when:
+        def pair = service.renameSoftInsert(child, newChild)
+        def child_parent_expected = [new ParentInfo(parent.toString(), type, parentUUID)] as Set
+        // Add a record with the same child name but different uuid and since this record is added after rename
+        // this record should not be dropped during the service.drop
+        insertNewParentChildRecord(parent.toString(), parentUUID, child.toString(), "c_uuid2", type)
+        service.drop(child, Optional.of(pair))
+        then:
+        // Test Parent
+        assert service.getParents(parent).isEmpty()
+        def parent_children_expected  = [new ChildInfo(newChild.toString(), type, childUUID),
+                                         new ChildInfo(child.toString(), type, "c_uuid2")] as Set
+        assert parent_children_expected == service.getChildren(parent)
+
+        // Test new child
+        assert service.getParents(newChild) == child_parent_expected
+        assert service.getChildren(newChild).isEmpty()
+
+        // Test old child
+        assert service.getParents(child) == child_parent_expected
+        assert service.getChildren(child).isEmpty()
+    }
+
+    // Add a test where between rename and drop old/new name same name but different uuid is added
+    // only those get from rename should be drop
+    // This case is not possible in reality but wanted to add a test to prove this
+    def "Simulate Rename Parent: drop only impacted uuids"() {
+        setup:
+        def parent = QualifiedName.ofTable(catalog, database, "p")
+        def parentUUID = "p_uuid"
+        def child = QualifiedName.ofTable(catalog, database, "c")
+        def childUUID = "c_uuid"
+        def type = "clone"
+        service.createParentChildRelation(parent, parentUUID, child, childUUID, type, props)
+        def newParent = QualifiedName.ofTable(catalog, database, "np")
+
+        when:
+        def pair = service.renameSoftInsert(parent, newParent)
+        // Add a record with the same parent name but different uuid and since this record is added after rename
+        // this record should not be dropped during the service.drop
+        insertNewParentChildRecord(parent.toString(), "p_uuid2", child.toString(), "c_uuid1", "clone")
+        service.drop(parent, Optional.of(pair))
+        then:
+
+        // Test p
+        assert service.getParents(parent).isEmpty()
+        def parent_children_expected  = [new ChildInfo(child.toString(), "clone", "c_uuid1")] as Set
+        assert parent_children_expected == service.getChildren(parent)
+
+        // Test np
+        assert service.getParents(newParent).isEmpty()
+        def new_parent_children_expected  = [new ChildInfo(child.toString(), "clone", childUUID)] as Set
+        assert new_parent_children_expected == service.getChildren(newParent)
+
+        // Test child
+        assert service.getChildren((child)).isEmpty()
+        def child_parent_expected = [
+            new ParentInfo(newParent.toString(), "clone", parentUUID),
+            new ParentInfo(parent.toString(), "clone", "p_uuid2"),
+        ] as Set
+        assert service.getParents(child) == child_parent_expected
+    }
+
 }