From ed6c05157f60328b0215bde544f7a4e9894fd15f Mon Sep 17 00:00:00 2001 From: lukegleeson Date: Thu, 18 Aug 2022 17:47:24 +0100 Subject: Performance Improvement: Batch Update DataNodes Implemented methods to perform a batch operation on updating datanodes Refactored replace data node(s) tree methods to update data node(s) and descendants Issue-ID: CPS-1203 Signed-off-by: lukegleeson Change-Id: I365d657422b19c9ce384110c9a23d041eaed06f4 --- .../spi/impl/CpsDataPersistenceServiceImpl.java | 48 +++++++++---- ...CpsDataPersistenceServiceIntegrationSpec.groovy | 42 +++++------ .../spi/impl/CpsDataPersistenceServiceSpec.groovy | 83 +++++++++++++++------- 3 files changed, 115 insertions(+), 58 deletions(-) (limited to 'cps-ri/src') diff --git a/cps-ri/src/main/java/org/onap/cps/spi/impl/CpsDataPersistenceServiceImpl.java b/cps-ri/src/main/java/org/onap/cps/spi/impl/CpsDataPersistenceServiceImpl.java index 944335598..c4a2c2fe9 100644 --- a/cps-ri/src/main/java/org/onap/cps/spi/impl/CpsDataPersistenceServiceImpl.java +++ b/cps-ri/src/main/java/org/onap/cps/spi/impl/CpsDataPersistenceServiceImpl.java @@ -26,6 +26,7 @@ import static org.onap.cps.spi.FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS; import com.google.common.collect.ImmutableSet; import com.google.common.collect.ImmutableSet.Builder; +import java.util.ArrayList; import java.util.Collection; import java.util.Collections; import java.util.HashMap; @@ -104,14 +105,16 @@ public class CpsDataPersistenceServiceImpl implements CpsDataPersistenceService final Collection newChildren) { final FragmentEntity parentFragmentEntity = getFragmentByXpath(dataspaceName, anchorName, parentNodeXpath); try { - for (final DataNode newChildAsDataNode : newChildren) { + final List fragmentEntities = new ArrayList<>(); + newChildren.forEach(newChildAsDataNode -> { final FragmentEntity newChildAsFragmentEntity = convertToFragmentWithAllDescendants( - parentFragmentEntity.getDataspace(), - parentFragmentEntity.getAnchor(), - newChildAsDataNode); + parentFragmentEntity.getDataspace(), + parentFragmentEntity.getAnchor(), + newChildAsDataNode); newChildAsFragmentEntity.setParentId(parentFragmentEntity.getId()); - fragmentRepository.save(newChildAsFragmentEntity); - } + fragmentEntities.add(newChildAsFragmentEntity); + }); + fragmentRepository.saveAll(fragmentEntities); } catch (final DataIntegrityViolationException exception) { final List conflictXpaths = newChildren.stream() .map(DataNode::getXpath) @@ -288,9 +291,10 @@ public class CpsDataPersistenceServiceImpl implements CpsDataPersistenceService } @Override - public void replaceDataNodeTree(final String dataspaceName, final String anchorName, final DataNode dataNode) { + public void updateDataNodeAndDescendants(final String dataspaceName, final String anchorName, + final DataNode dataNode) { final FragmentEntity fragmentEntity = getFragmentByXpath(dataspaceName, anchorName, dataNode.getXpath()); - replaceDataNodeTree(fragmentEntity, dataNode); + updateFragmentEntityAndDescendantsWithDataNode(fragmentEntity, dataNode); try { fragmentRepository.save(fragmentEntity); } catch (final StaleStateException staleStateException) { @@ -301,8 +305,27 @@ public class CpsDataPersistenceServiceImpl implements CpsDataPersistenceService } } - private void replaceDataNodeTree(final FragmentEntity existingFragmentEntity, - final DataNode newDataNode) { + @Override + public void updateDataNodesAndDescendants(final String dataspaceName, + final String anchorName, + final List dataNodes) { + final Map dataNodeFragmentEntityMap = dataNodes.stream() + .collect(Collectors.toMap( + dataNode -> dataNode, dataNode -> getFragmentByXpath(dataspaceName, anchorName, dataNode.getXpath()))); + dataNodeFragmentEntityMap.forEach( + (dataNode, fragmentEntity) -> updateFragmentEntityAndDescendantsWithDataNode(fragmentEntity, dataNode)); + try { + fragmentRepository.saveAll(dataNodeFragmentEntityMap.values()); + } catch (final StaleStateException staleStateException) { + throw new ConcurrencyException("Concurrent Transactions", + String.format("A data node in dataspace :'%s' with Anchor : '%s' is updated by another transaction.", + dataspaceName, anchorName), + staleStateException); + } + } + + private void updateFragmentEntityAndDescendantsWithDataNode(final FragmentEntity existingFragmentEntity, + final DataNode newDataNode) { existingFragmentEntity.setAttributes(jsonObjectMapper.asJsonString(newDataNode.getLeaves())); @@ -318,10 +341,11 @@ public class CpsDataPersistenceServiceImpl implements CpsDataPersistenceService existingFragmentEntity.getDataspace(), existingFragmentEntity.getAnchor(), newDataNodeChild); } else { childFragment = existingChildrenByXpath.get(newDataNodeChild.getXpath()); - replaceDataNodeTree(childFragment, newDataNodeChild); + updateFragmentEntityAndDescendantsWithDataNode(childFragment, newDataNodeChild); } updatedChildFragments.add(childFragment); } + existingFragmentEntity.getChildFragments().clear(); existingFragmentEntity.getChildFragments().addAll(updatedChildFragments); } @@ -457,7 +481,7 @@ public class CpsDataPersistenceServiceImpl implements CpsDataPersistenceService copyAttributesFromNewListElement(existingListElementEntity, newListElement); existingListElementEntity.getChildFragments().clear(); } else { - replaceDataNodeTree(existingListElementEntity, newListElement); + updateFragmentEntityAndDescendantsWithDataNode(existingListElementEntity, newListElement); } return existingListElementEntity; } diff --git a/cps-ri/src/test/groovy/org/onap/cps/spi/impl/CpsDataPersistenceServiceIntegrationSpec.groovy b/cps-ri/src/test/groovy/org/onap/cps/spi/impl/CpsDataPersistenceServiceIntegrationSpec.groovy index 6f780fc50..fee489d18 100755 --- a/cps-ri/src/test/groovy/org/onap/cps/spi/impl/CpsDataPersistenceServiceIntegrationSpec.groovy +++ b/cps-ri/src/test/groovy/org/onap/cps/spi/impl/CpsDataPersistenceServiceIntegrationSpec.groovy @@ -55,8 +55,8 @@ class CpsDataPersistenceServiceIntegrationSpec extends CpsPersistenceSpecBase { static final long ID_DATA_NODE_WITH_DESCENDANTS = 4001 static final String XPATH_DATA_NODE_WITH_DESCENDANTS = '/parent-1' static final String XPATH_DATA_NODE_WITH_LEAVES = '/parent-100' - static final long UPDATE_DATA_NODE_FRAGMENT_ID = 4202L - static final long UPDATE_DATA_NODE_SUB_FRAGMENT_ID = 4203L + static final long DATA_NODE_202_FRAGMENT_ID = 4202L + static final long CHILD_OF_DATA_NODE_202_FRAGMENT_ID = 4203L static final long LIST_DATA_NODE_PARENT201_FRAGMENT_ID = 4206L static final long LIST_DATA_NODE_PARENT203_FRAGMENT_ID = 4214L static final long LIST_DATA_NODE_PARENT202_FRAGMENT_ID = 4211L @@ -258,14 +258,14 @@ class CpsDataPersistenceServiceIntegrationSpec extends CpsPersistenceSpecBase { objectUnderTest.updateDataLeaves(DATASPACE_NAME, ANCHOR_FOR_DATA_NODES_WITH_LEAVES, "/parent-200/child-201", ['leaf-value': 'new']) then: 'leaves are updated for selected data node' - def updatedFragment = fragmentRepository.getById(UPDATE_DATA_NODE_FRAGMENT_ID) + def updatedFragment = fragmentRepository.getById(DATA_NODE_202_FRAGMENT_ID) def updatedLeaves = getLeavesMap(updatedFragment) assert updatedLeaves.size() == 1 assert updatedLeaves.'leaf-value' == 'new' and: 'existing child entry remains as is' def childFragment = updatedFragment.childFragments.iterator().next() def childLeaves = getLeavesMap(childFragment) - assert childFragment.id == UPDATE_DATA_NODE_SUB_FRAGMENT_ID + assert childFragment.id == CHILD_OF_DATA_NODE_202_FRAGMENT_ID assert childLeaves.'leaf-value' == 'original' } @@ -283,32 +283,32 @@ class CpsDataPersistenceServiceIntegrationSpec extends CpsPersistenceSpecBase { } @Sql([CLEAR_DATA, SET_DATA]) - def 'Replace data node tree with descendants removal.'() { + def 'Update data node and descendants by removing descendants.'() { given: 'data node object with leaves updated, no children' def submittedDataNode = buildDataNode("/parent-200/child-201", ['leaf-value': 'new'], []) - when: 'replace data node tree is performed' - objectUnderTest.replaceDataNodeTree(DATASPACE_NAME, ANCHOR_FOR_DATA_NODES_WITH_LEAVES, submittedDataNode) + when: 'update data nodes and descendants is performed' + objectUnderTest.updateDataNodeAndDescendants(DATASPACE_NAME, ANCHOR_FOR_DATA_NODES_WITH_LEAVES, submittedDataNode) then: 'leaves have been updated for selected data node' - def updatedFragment = fragmentRepository.getById(UPDATE_DATA_NODE_FRAGMENT_ID) + def updatedFragment = fragmentRepository.getById(DATA_NODE_202_FRAGMENT_ID) def updatedLeaves = getLeavesMap(updatedFragment) assert updatedLeaves.size() == 1 assert updatedLeaves.'leaf-value' == 'new' and: 'updated entry has no children' updatedFragment.childFragments.isEmpty() and: 'previously attached child entry is removed from database' - fragmentRepository.findById(UPDATE_DATA_NODE_SUB_FRAGMENT_ID).isEmpty() + fragmentRepository.findById(CHILD_OF_DATA_NODE_202_FRAGMENT_ID).isEmpty() } @Sql([CLEAR_DATA, SET_DATA]) - def 'Replace data node tree with descendants.'() { + def 'Update data node and descendants with new descendants'() { given: 'data node object with leaves updated, having child with old content' def submittedDataNode = buildDataNode("/parent-200/child-201", ['leaf-value': 'new'], [ buildDataNode("/parent-200/child-201/grand-child", ['leaf-value': 'original'], []) ]) when: 'update is performed including descendants' - objectUnderTest.replaceDataNodeTree(DATASPACE_NAME, ANCHOR_FOR_DATA_NODES_WITH_LEAVES, submittedDataNode) + objectUnderTest.updateDataNodeAndDescendants(DATASPACE_NAME, ANCHOR_FOR_DATA_NODES_WITH_LEAVES, submittedDataNode) then: 'leaves have been updated for selected data node' - def updatedFragment = fragmentRepository.getById(UPDATE_DATA_NODE_FRAGMENT_ID) + def updatedFragment = fragmentRepository.getById(DATA_NODE_202_FRAGMENT_ID) def updatedLeaves = getLeavesMap(updatedFragment) assert updatedLeaves.size() == 1 assert updatedLeaves.'leaf-value' == 'new' @@ -320,15 +320,15 @@ class CpsDataPersistenceServiceIntegrationSpec extends CpsPersistenceSpecBase { } @Sql([CLEAR_DATA, SET_DATA]) - def 'Replace data node tree with same descendants but changed leaf value.'() { + def 'Update data node and descendants with same descendants but changed leaf value.'() { given: 'data node object with leaves updated, having child with old content' def submittedDataNode = buildDataNode("/parent-200/child-201", ['leaf-value': 'new'], [ buildDataNode("/parent-200/child-201/grand-child", ['leaf-value': 'new'], []) ]) when: 'update is performed including descendants' - objectUnderTest.replaceDataNodeTree(DATASPACE_NAME, ANCHOR_FOR_DATA_NODES_WITH_LEAVES, submittedDataNode) + objectUnderTest.updateDataNodeAndDescendants(DATASPACE_NAME, ANCHOR_FOR_DATA_NODES_WITH_LEAVES, submittedDataNode) then: 'leaves have been updated for selected data node' - def updatedFragment = fragmentRepository.getById(UPDATE_DATA_NODE_FRAGMENT_ID) + def updatedFragment = fragmentRepository.getById(DATA_NODE_202_FRAGMENT_ID) def updatedLeaves = getLeavesMap(updatedFragment) assert updatedLeaves.size() == 1 assert updatedLeaves.'leaf-value' == 'new' @@ -340,20 +340,20 @@ class CpsDataPersistenceServiceIntegrationSpec extends CpsPersistenceSpecBase { } @Sql([CLEAR_DATA, SET_DATA]) - def 'Replace data node tree with different descendants xpath'() { + def 'Update data node and descendants with different descendants xpath'() { given: 'data node object with leaves updated, having child with old content' def submittedDataNode = buildDataNode("/parent-200/child-201", ['leaf-value': 'new'], [ buildDataNode("/parent-200/child-201/grand-child-new", ['leaf-value': 'new'], []) ]) when: 'update is performed including descendants' - objectUnderTest.replaceDataNodeTree(DATASPACE_NAME, ANCHOR_FOR_DATA_NODES_WITH_LEAVES, submittedDataNode) + objectUnderTest.updateDataNodeAndDescendants(DATASPACE_NAME, ANCHOR_FOR_DATA_NODES_WITH_LEAVES, submittedDataNode) then: 'leaves have been updated for selected data node' - def updatedFragment = fragmentRepository.getById(UPDATE_DATA_NODE_FRAGMENT_ID) + def updatedFragment = fragmentRepository.getById(DATA_NODE_202_FRAGMENT_ID) def updatedLeaves = getLeavesMap(updatedFragment) assert updatedLeaves.size() == 1 assert updatedLeaves.'leaf-value' == 'new' and: 'previously attached child entry is removed from database' - fragmentRepository.findById(UPDATE_DATA_NODE_SUB_FRAGMENT_ID).isEmpty() + fragmentRepository.findById(CHILD_OF_DATA_NODE_202_FRAGMENT_ID).isEmpty() and: 'new child entry is persisted' def childFragment = updatedFragment.childFragments.iterator().next() childFragment.xpath == '/parent-200/child-201/grand-child-new' @@ -362,11 +362,11 @@ class CpsDataPersistenceServiceIntegrationSpec extends CpsPersistenceSpecBase { } @Sql([CLEAR_DATA, SET_DATA]) - def 'Replace data node tree error scenario: #scenario.'() { + def 'Update data node and descendants error scenario: #scenario.'() { given: 'data node object' def submittedDataNode = buildDataNode(xpath, ['leaf-name': 'leaf-value'], []) when: 'attempt to update data node for #scenario' - objectUnderTest.replaceDataNodeTree(dataspaceName, anchorName, submittedDataNode) + objectUnderTest.updateDataNodeAndDescendants(dataspaceName, anchorName, submittedDataNode) then: 'a #expectedException is thrown' thrown(expectedException) where: 'the following data is used' diff --git a/cps-ri/src/test/groovy/org/onap/cps/spi/impl/CpsDataPersistenceServiceSpec.groovy b/cps-ri/src/test/groovy/org/onap/cps/spi/impl/CpsDataPersistenceServiceSpec.groovy index bde2f3de9..1bbf358e5 100644 --- a/cps-ri/src/test/groovy/org/onap/cps/spi/impl/CpsDataPersistenceServiceSpec.groovy +++ b/cps-ri/src/test/groovy/org/onap/cps/spi/impl/CpsDataPersistenceServiceSpec.groovy @@ -28,6 +28,7 @@ import org.onap.cps.spi.entities.SchemaSetEntity import org.onap.cps.spi.entities.YangResourceEntity import org.onap.cps.spi.exceptions.ConcurrencyException import org.onap.cps.spi.exceptions.DataValidationException +import org.onap.cps.spi.model.DataNode import org.onap.cps.spi.model.DataNodeBuilder import org.onap.cps.spi.repository.AnchorRepository import org.onap.cps.spi.repository.DataspaceRepository @@ -67,40 +68,40 @@ class CpsDataPersistenceServiceSpec extends Specification { )] as Set - def 'Handling of StaleStateException (caused by concurrent updates) during data node tree update.'() { - - def parentXpath = '/parent-01' - def myDataspaceName = 'my-dataspace' - def myAnchorName = 'my-anchor' - - given: 'data node object' - def submittedDataNode = new DataNodeBuilder() - .withXpath(parentXpath) - .withLeaves(['leaf-name': 'leaf-value']) - .build() - and: 'fragment to be updated' - mockFragmentRepository.getByDataspaceAndAnchorAndXpath(_, _, _) >> { + def 'Handling of StaleStateException (caused by concurrent updates) during update data node and descendants.'() { + given: 'the fragment repository returns a fragment entity' + mockFragmentRepository.getByDataspaceAndAnchorAndXpath(*_) >> { def fragmentEntity = new FragmentEntity() - fragmentEntity.setXpath(parentXpath) - fragmentEntity.setChildFragments(Collections.emptySet()) + fragmentEntity.setChildFragments([new FragmentEntity()] as Set) return fragmentEntity } - and: 'data node is concurrently updated by another transaction' + and: 'a data node is concurrently updated by another transaction' mockFragmentRepository.save(_) >> { throw new StaleStateException("concurrent updates") } + when: 'attempt to update data node with submitted data nodes' + objectUnderTest.updateDataNodeAndDescendants('some-dataspace', 'some-anchor', new DataNodeBuilder().withXpath('/some/xpath').build()) + then: 'concurrency exception is thrown' + def concurrencyException = thrown(ConcurrencyException) + assert concurrencyException.getDetails().contains('some-dataspace') + assert concurrencyException.getDetails().contains('some-anchor') + assert concurrencyException.getDetails().contains('/some/xpath') + } - when: 'attempt to update data node' - objectUnderTest.replaceDataNodeTree(myDataspaceName, myAnchorName, submittedDataNode) - + def 'Handling of StaleStateException (caused by concurrent updates) during update data nodes and descendants.'() { + given: 'the fragment repository returns a list of fragment entities' + mockFragmentRepository.getByDataspaceAndAnchorAndXpath(*_) >> new FragmentEntity() + and: 'a data node is concurrently updated by another transaction' + mockFragmentRepository.saveAll(*_) >> { throw new StaleStateException("concurrent updates") } + when: 'attempt to update data node with submitted data nodes' + objectUnderTest.updateDataNodesAndDescendants('some-dataspace', 'some-anchor', []) then: 'concurrency exception is thrown' def concurrencyException = thrown(ConcurrencyException) - assert concurrencyException.getDetails().contains(myDataspaceName) - assert concurrencyException.getDetails().contains(myAnchorName) - assert concurrencyException.getDetails().contains(parentXpath) + assert concurrencyException.getDetails().contains('some-dataspace') + assert concurrencyException.getDetails().contains('some-anchor') } def 'Retrieving a data node with a property JSON value of #scenario'() { given: 'a fragment with a property JSON value of #scenario' - mockFragmentRepository.getByDataspaceAndAnchorAndXpath(_, _, _) >> { + mockFragmentRepository.getByDataspaceAndAnchorAndXpath(*_) >> { new FragmentEntity(childFragments: Collections.emptySet(), attributes: "{\"some attribute\": ${dataString}}", anchor: new AnchorEntity(schemaSet: new SchemaSetEntity(yangResources: yangResourceSet ))) @@ -128,11 +129,11 @@ class CpsDataPersistenceServiceSpec extends Specification { def 'Retrieving a data node with invalid JSON'() { given: 'a fragment with invalid JSON' - mockFragmentRepository.getByDataspaceAndAnchorAndXpath(_, _, _) >> { + mockFragmentRepository.getByDataspaceAndAnchorAndXpath(*_) >> { new FragmentEntity(childFragments: Collections.emptySet(), attributes: '{invalid json') } when: 'getting the data node represented by this fragment' - def dataNode = objectUnderTest.getDataNode('my-dataspace', 'my-anchor', + objectUnderTest.getDataNode('my-dataspace', 'my-anchor', '/parent-01', FetchDescendantsOption.INCLUDE_ALL_DESCENDANTS) then: 'a data validation exception is thrown' thrown(DataValidationException) @@ -160,4 +161,36 @@ class CpsDataPersistenceServiceSpec extends Specification { then: 'the session manager method to lock anchor is invoked with same parameters' 1 * mockSessionManager.lockAnchor('mySessionId', 'myDataspaceName', 'myAnchorName', 123L) } + + def 'update data node and descendants: #scenario'(){ + given: 'mocked responses' + mockFragmentRepository.getByDataspaceAndAnchorAndXpath(_, _, '/test/xpath') >> new FragmentEntity(xpath: '/test/xpath', childFragments: []) + when: 'replace data node tree' + objectUnderTest.updateDataNodesAndDescendants('dataspaceName', 'anchorName', dataNodes) + then: 'call fragment repository save all method' + 1 * mockFragmentRepository.saveAll({fragmentEntities -> assert fragmentEntities as List == expectedFragmentEntities}) + where: 'the following Data Type is passed' + scenario | dataNodes || expectedFragmentEntities + 'empty data node list' | [] || [] + 'one data node in list' | [new DataNode(xpath: '/test/xpath', leaves: ['id': 'testId'], childDataNodes: [])] || [new FragmentEntity(xpath: '/test/xpath', attributes: '{"id":"testId"}', childFragments: [])] + } + + def 'update data nodes and descendants'() { + given: 'the fragment repository returns a fragment entity related to the xpath input' + mockFragmentRepository.getByDataspaceAndAnchorAndXpath(_, _, '/test/xpath1') >> new FragmentEntity(xpath: '/test/xpath1', childFragments: []) + mockFragmentRepository.getByDataspaceAndAnchorAndXpath(_, _, '/test/xpath2') >> new FragmentEntity(xpath: '/test/xpath2', childFragments: []) + and: 'some data nodes with descendants' + def dataNode1 = new DataNode(xpath: '/test/xpath1', leaves: ['id': 'testId1'], childDataNodes: [new DataNode(xpath: '/test/xpath1/child', leaves: ['id': 'childTestId1'])]) + def dataNode2 = new DataNode(xpath: '/test/xpath2', leaves: ['id': 'testId2'], childDataNodes: [new DataNode(xpath: '/test/xpath2/child', leaves: ['id': 'childTestId2'])]) + when: 'the fragment entities are update by the data nodes' + objectUnderTest.updateDataNodesAndDescendants('dataspaceName', 'anchorName', [dataNode1, dataNode2]) + then: 'call fragment repository save all method is called with the updated fragments' + 1 * mockFragmentRepository.saveAll({fragmentEntities -> { + fragmentEntities.containsAll([ + new FragmentEntity(xpath: '/test/xpath1', attributes: '{"id":"testId1"}', childFragments: [new FragmentEntity(xpath: '/test/xpath1/child', attributes: '{"id":"childTestId1"}', childFragments: [])]), + new FragmentEntity(xpath: '/test/xpath2', attributes: '{"id":"testId2"}', childFragments: [new FragmentEntity(xpath: '/test/xpath2/child', attributes: '{"id":"childTestId2"}', childFragments: [])]) + ]) + assert fragmentEntities.size() == 2 + }}) + } } \ No newline at end of file -- cgit 1.2.3-korg