http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/71b5b1c9/extras/rya.merger/src/test/java/mvm/rya/accumulo/mr/merge/MergeToolTest.java ---------------------------------------------------------------------- diff --git a/extras/rya.merger/src/test/java/mvm/rya/accumulo/mr/merge/MergeToolTest.java b/extras/rya.merger/src/test/java/mvm/rya/accumulo/mr/merge/MergeToolTest.java new file mode 100644 index 0000000..6203127 --- /dev/null +++ b/extras/rya.merger/src/test/java/mvm/rya/accumulo/mr/merge/MergeToolTest.java @@ -0,0 +1,475 @@ +package mvm.rya.accumulo.mr.merge; + +import static mvm.rya.accumulo.mr.merge.util.TestUtils.LAST_MONTH; +import static mvm.rya.accumulo.mr.merge.util.TestUtils.TODAY; +import static mvm.rya.accumulo.mr.merge.util.TestUtils.YESTERDAY; +import static mvm.rya.accumulo.mr.merge.util.TestUtils.createRyaStatement; +import static mvm.rya.accumulo.mr.merge.util.ToolConfigUtils.makeArgument; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.fail; + +import java.util.Date; +import java.util.Iterator; +import java.util.Map.Entry; +import java.util.TreeSet; + +import org.apache.accumulo.core.client.AccumuloSecurityException; +import org.apache.accumulo.core.client.Connector; +import org.apache.accumulo.core.client.Scanner; +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.Value; +import org.apache.accumulo.core.security.Authorizations; +import org.apache.accumulo.core.security.ColumnVisibility; +import org.apache.hadoop.io.Text; +import org.apache.log4j.Logger; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +import info.aduna.iteration.CloseableIteration; +import mvm.rya.accumulo.AccumuloRdfConfiguration; +import mvm.rya.accumulo.AccumuloRyaDAO; +import mvm.rya.accumulo.mr.merge.driver.AccumuloDualInstanceDriver; +import mvm.rya.accumulo.mr.merge.util.AccumuloInstanceDriver; +import mvm.rya.accumulo.mr.merge.util.AccumuloRyaUtils; +import mvm.rya.accumulo.mr.merge.util.TestUtils; +import mvm.rya.accumulo.mr.merge.util.TimeUtils; +import mvm.rya.accumulo.mr.utils.MRUtils; +import mvm.rya.api.RdfCloudTripleStoreConfiguration; +import mvm.rya.api.RdfCloudTripleStoreConstants; +import mvm.rya.api.domain.RyaStatement; +import mvm.rya.api.persist.RyaDAOException; + +/** + * Tests for {@link MergeTool}. + */ +public class MergeToolTest { + private static final Logger log = Logger.getLogger(MergeToolTest.class); + + private static final boolean IS_MOCK = true; + private static final boolean USE_TIME_SYNC = false; + private static final boolean IS_START_TIME_DIALOG_ENABLED = false; + + private static final String CHILD_SUFFIX = MergeTool.CHILD_SUFFIX; + + private static final String PARENT_PASSWORD = AccumuloDualInstanceDriver.PARENT_PASSWORD; + private static final String PARENT_INSTANCE = AccumuloDualInstanceDriver.PARENT_INSTANCE; + private static final String PARENT_TABLE_PREFIX = AccumuloDualInstanceDriver.PARENT_TABLE_PREFIX; + private static final String PARENT_AUTH = AccumuloDualInstanceDriver.PARENT_AUTH; + private static final ColumnVisibility PARENT_COLUMN_VISIBILITY = new ColumnVisibility(PARENT_AUTH); + private static final String PARENT_TOMCAT_URL = "http://rya-example-box:8080";; + + private static final String CHILD_PASSWORD = AccumuloDualInstanceDriver.CHILD_PASSWORD; + private static final String CHILD_INSTANCE = AccumuloDualInstanceDriver.CHILD_INSTANCE; + private static final String CHILD_TABLE_PREFIX = AccumuloDualInstanceDriver.CHILD_TABLE_PREFIX; + private static final String CHILD_AUTH = AccumuloDualInstanceDriver.CHILD_AUTH; + private static final ColumnVisibility CHILD_COLUMN_VISIBILITY = new ColumnVisibility(CHILD_AUTH); + private static final String CHILD_TOMCAT_URL = "http://localhost:8080";; + + private static Connector parentConnector; + private static Connector childConnector; + + private static AccumuloRyaDAO parentDao; + private static AccumuloRyaDAO childDao; + + private static AccumuloRdfConfiguration parentConfig; + private static AccumuloRdfConfiguration childConfig; + + private static AccumuloDualInstanceDriver accumuloDualInstanceDriver; + + @BeforeClass + public static void setUp() throws Exception { + accumuloDualInstanceDriver = new AccumuloDualInstanceDriver(IS_MOCK, false, false, true, true); + accumuloDualInstanceDriver.setUpInstances(); + + parentConnector = accumuloDualInstanceDriver.getParentConnector(); + childConnector = accumuloDualInstanceDriver.getChildConnector(); + } + + @Before + public void setUpPerTest() throws Exception { + accumuloDualInstanceDriver.setUpTables(); + + accumuloDualInstanceDriver.setUpDaos(); + + accumuloDualInstanceDriver.setUpConfigs(); + + parentConfig = accumuloDualInstanceDriver.getParentConfig(); + childConfig = accumuloDualInstanceDriver.getChildConfig(); + parentDao = accumuloDualInstanceDriver.getParentDao(); + childDao = accumuloDualInstanceDriver.getChildDao(); + } + + @After + public void tearDownPerTest() throws Exception { + log.info("tearDownPerTest(): tearing down now."); + accumuloDualInstanceDriver.tearDownTables(); + accumuloDualInstanceDriver.tearDownDaos(); + } + + @AfterClass + public static void tearDownPerClass() throws Exception { + log.info("tearDownPerClass(): tearing down now."); + accumuloDualInstanceDriver.tearDown(); + } + + private void assertStatementInParent(String description, int verifyResultCount, RyaStatement matchStatement) throws RyaDAOException { + TestUtils.assertStatementInInstance(description, verifyResultCount, matchStatement, parentDao, parentConfig); + } + + private void mergeToolRun(Date startDate) { + MergeTool.setupAndRun(new String[] { + makeArgument(MRUtils.AC_MOCK_PROP, Boolean.toString(IS_MOCK)), + makeArgument(MRUtils.AC_INSTANCE_PROP, PARENT_INSTANCE), + makeArgument(MRUtils.AC_USERNAME_PROP, accumuloDualInstanceDriver.getParentUser()), + makeArgument(MRUtils.AC_PWD_PROP, PARENT_PASSWORD), + makeArgument(MRUtils.TABLE_PREFIX_PROPERTY, PARENT_TABLE_PREFIX), + makeArgument(MRUtils.AC_AUTH_PROP, PARENT_AUTH), + makeArgument(MRUtils.AC_ZK_PROP, accumuloDualInstanceDriver.getParentZooKeepers()), + makeArgument(CopyTool.PARENT_TOMCAT_URL_PROP, PARENT_TOMCAT_URL), + makeArgument(MRUtils.AC_MOCK_PROP + CHILD_SUFFIX, Boolean.toString(IS_MOCK)), + makeArgument(MRUtils.AC_INSTANCE_PROP + CHILD_SUFFIX, CHILD_INSTANCE), + makeArgument(MRUtils.AC_USERNAME_PROP + CHILD_SUFFIX, accumuloDualInstanceDriver.getChildUser()), + makeArgument(MRUtils.AC_PWD_PROP + CHILD_SUFFIX, CHILD_PASSWORD), + makeArgument(MRUtils.TABLE_PREFIX_PROPERTY + CHILD_SUFFIX, CHILD_TABLE_PREFIX), + makeArgument(MRUtils.AC_AUTH_PROP + CHILD_SUFFIX, CHILD_AUTH), + makeArgument(MRUtils.AC_ZK_PROP + CHILD_SUFFIX, accumuloDualInstanceDriver.getChildZooKeepers()), + makeArgument(CopyTool.CHILD_TOMCAT_URL_PROP, CHILD_TOMCAT_URL), + makeArgument(CopyTool.NTP_SERVER_HOST_PROP, TimeUtils.DEFAULT_TIME_SERVER_HOST), + makeArgument(CopyTool.USE_NTP_SERVER_PROP, Boolean.toString(USE_TIME_SYNC)), + makeArgument(MergeTool.START_TIME_PROP, MergeTool.getStartTimeString(startDate, IS_START_TIME_DIALOG_ENABLED)) + }); + + log.info("Finished running tool."); + } + + @Test + public void testMergeTool() throws Exception { + // This statement was in both parent/child instances a month ago and is before the start time of yesterday + // but it was left alone. It should remain in the parent after merging. + RyaStatement ryaStatementOutOfTimeRange = createRyaStatement("coach", "called", "timeout", LAST_MONTH); + + // This statement was in both parent/child instances a month ago but after the start time of yesterday + // the parent deleted it and the child still has it. It should stay deleted in the parent after merging. + RyaStatement ryaStatementParentDeletedAfter = createRyaStatement("parent", "deleted", "after", LAST_MONTH); + + // This statement was added by the parent after the start time of yesterday and doesn't exist in the child. + // It should stay in the parent after merging. + RyaStatement ryaStatementParentAddedAfter = createRyaStatement("parent", "added", "after", TODAY); + + // This statement was in both parent/child instances a month ago but after the start time of yesterday + // the child deleted it and the parent still has it. It should be deleted from the parent after merging. + RyaStatement ryaStatementChildDeletedAfter = createRyaStatement("child", "deleted", "after", LAST_MONTH); + + // This statement was added by the child after the start time of yesterday and doesn't exist in the parent. + // It should be added to the parent after merging. + RyaStatement ryaStatementChildAddedAfter = createRyaStatement("child", "added", "after", TODAY); + + // This statement was modified by the child after the start of yesterday (The timestamp changes after updating) + // It should be updated in the parent to match the child. + RyaStatement ryaStatementUpdatedByChild = createRyaStatement("bob", "catches", "ball", LAST_MONTH); + + RyaStatement ryaStatementUntouchedByChild = createRyaStatement("bill", "talks to", "john", LAST_MONTH); + + RyaStatement ryaStatementDeletedByChild = createRyaStatement("susan", "eats", "burgers", LAST_MONTH); + + RyaStatement ryaStatementAddedByChild = createRyaStatement("ronnie", "plays", "guitar", TODAY); + + // This statement was modified by the child to change the column visibility. + // The parent should combine the child's visibility with its visibility. + RyaStatement ryaStatementVisibilityDifferent = createRyaStatement("I", "see", "you", LAST_MONTH); + ryaStatementVisibilityDifferent.setColumnVisibility(PARENT_COLUMN_VISIBILITY.getExpression()); + + // Setup initial parent instance with 7 rows + // This is the state of the parent data (as it is today) before merging occurs which will use the specified start time of yesterday. + parentDao.add(ryaStatementOutOfTimeRange); // Merging should keep statement + parentDao.add(ryaStatementUpdatedByChild); // Merging should update statement + parentDao.add(ryaStatementUntouchedByChild); // Merging should keep statement + parentDao.add(ryaStatementDeletedByChild); // Merging should delete statement + parentDao.add(ryaStatementVisibilityDifferent); // Merging should update statement + parentDao.add(ryaStatementParentAddedAfter); // Merging should keep statement + parentDao.add(ryaStatementChildDeletedAfter); // Merging should delete statement + + // Simulate the child coming back with a modified data set before the merging occurs. + // (1 updated row, 1 row left alone because it was unchanged, 1 row outside time range, + // 1 row deleted, 1 new row added, 1 modified visibility, 1 deleted by child, 1 added by child). + // There should be 5 rows in the child instance (4 which will be scanned over from the start time). + ryaStatementUpdatedByChild.setObject(TestUtils.createRyaUri("football")); + ryaStatementUpdatedByChild.setTimestamp(TODAY.getTime()); + ryaStatementVisibilityDifferent.setColumnVisibility(CHILD_COLUMN_VISIBILITY.getExpression()); + childDao.add(ryaStatementOutOfTimeRange); + childDao.add(ryaStatementUpdatedByChild); + childDao.add(ryaStatementUntouchedByChild); + childDao.add(ryaStatementAddedByChild); // Merging should add statement + childDao.add(ryaStatementVisibilityDifferent); + childDao.add(ryaStatementParentDeletedAfter); + childDao.add(ryaStatementChildAddedAfter); // Merging should add statement + + AccumuloRyaUtils.printTable(PARENT_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, parentConfig); + AccumuloRyaUtils.printTable(CHILD_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, childConfig); + + log.info("Starting merge tool. Merging all data after the specified start time: " + YESTERDAY); + + mergeToolRun(YESTERDAY); + + + for (String tableSuffix : AccumuloInstanceDriver.TABLE_NAME_SUFFIXES) { + AccumuloRyaUtils.printTable(PARENT_TABLE_PREFIX + tableSuffix, parentConfig); + } + + Scanner scanner = AccumuloRyaUtils.getScanner(PARENT_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, parentConfig); + Iterator<Entry<Key, Value>> iterator = scanner.iterator(); + int count = 0; + while (iterator.hasNext()) { + iterator.next(); + count++; + } + // Make sure we have all of them in the parent. + assertEquals(7, count); + + + assertStatementInParent("Parent missing statement that untouched by the child", 1, ryaStatementUntouchedByChild); + + assertStatementInParent("Parent missing statement that was out of time range", 1, ryaStatementOutOfTimeRange); + + assertStatementInParent("Parent missing statement that was updated by the child", 1, ryaStatementUpdatedByChild); + + assertStatementInParent("Parent missing statement that was added by the child", 1, ryaStatementAddedByChild); + + assertStatementInParent("Parent has statement that the child deleted", 0, ryaStatementDeletedByChild); + + // Check that it can be queried with parent's visibility + assertStatementInParent("Parent missing statement with parent visibility", 1, ryaStatementVisibilityDifferent); + + // Check that it can be queried with child's visibility + parentConfig.set(RdfCloudTripleStoreConfiguration.CONF_QUERY_AUTH, CHILD_AUTH); + Authorizations newParentAuths = AccumuloRyaUtils.addUserAuths(accumuloDualInstanceDriver.getParentUser(), accumuloDualInstanceDriver.getParentSecOps(), CHILD_AUTH); + accumuloDualInstanceDriver.getParentSecOps().changeUserAuthorizations(accumuloDualInstanceDriver.getParentUser(), newParentAuths); + assertStatementInParent("Parent missing statement with child visibility", 1, ryaStatementVisibilityDifferent); + + // Check that it can NOT be queried with some other visibility + parentConfig.set(RdfCloudTripleStoreConfiguration.CONF_QUERY_AUTH, "bad_auth"); + CloseableIteration<RyaStatement, RyaDAOException> iter = parentDao.getQueryEngine().query(ryaStatementVisibilityDifferent, parentConfig); + count = 0; + try { + while (iter.hasNext()) { + iter.next(); + count++; + } + } catch (Exception e) { + // Expected + if (!(e.getCause() instanceof AccumuloSecurityException)) { + fail(); + } + } + iter.close(); + assertEquals(0, count); + + // reset auth + parentConfig.set(RdfCloudTripleStoreConfiguration.CONF_QUERY_AUTH, PARENT_AUTH); + + assertStatementInParent("Parent has statement it deleted later", 0, ryaStatementParentDeletedAfter); + + assertStatementInParent("Parent missing statement it added later", 1, ryaStatementParentAddedAfter); + + assertStatementInParent("Parent has statement child deleted later", 0, ryaStatementChildDeletedAfter); + + assertStatementInParent("Parent missing statement child added later", 1, ryaStatementChildAddedAfter); + + + log.info("DONE"); + } + + private static RyaStatement createRyaStatementUnique(String s, String p, String o, Date date) throws Exception { + String uniquePart = Long.toString(System.currentTimeMillis() & 0xffffff, 64); + return createRyaStatement(s+uniquePart, p+uniquePart, o+uniquePart, date); + } + + private static RyaStatement createRyaStatementUniqueAdd(String s, String p, String o, Date date, AccumuloRyaDAO dao1, AccumuloRyaDAO dao2) throws Exception { + String uniquePart = Long.toString(System.currentTimeMillis() & 0xffffff, 64); + RyaStatement rs = createRyaStatement(s + uniquePart, p + uniquePart, o + uniquePart, date); + if (dao1 != null) { + dao1.add(rs); + } + if (dao2 != null) { + dao2.add(rs); + } + return rs; + } + + @Test + public void testMissingParentNewChild() throws Exception { + RyaStatement stmtNewInChild = createRyaStatementUnique("s_newInChild", "p_newInChild", "o_newInChild", null); + RyaStatement stmtSameInBoth = createRyaStatementUnique("s_same", "p_same", "o_same", LAST_MONTH); + childDao.add(stmtNewInChild); // Merging should add statement to parent + childDao.add(stmtSameInBoth); // Merging should ignore statement + parentDao.add(stmtSameInBoth); // Merging should ignore statement + mergeToolRun(YESTERDAY); + assertStatementInParent("new child statement added in parent ", 1, stmtNewInChild); + assertStatementInParent("Statement in p and child. ", 1, stmtSameInBoth); // Merging should ignore statement + } + + @Test + public void testOldParentMissingChild() throws Exception { + RyaStatement stmtMissingInChildOld = createRyaStatementUniqueAdd("s_notInChild", "p_notInChild", "o_notInChild", LAST_MONTH, parentDao, null); + mergeToolRun(YESTERDAY); + assertStatementInParent("Missing in child statement deleted old in parent ", 0, stmtMissingInChildOld); + } + + @Test + public void testNewParentEmptyChild() throws Exception { + RyaStatement stmtNewP_MisC = createRyaStatementUniqueAdd("s_NewP_MisC", "p_NewP_MisC", "o_NewP_MisC", null, parentDao, null); + AccumuloRyaUtils.printTable(PARENT_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, parentConfig); + AccumuloRyaUtils.printTable(CHILD_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, childConfig); + mergeToolRun(YESTERDAY); + // Note that nothing changes. This should catch issues with empty tables. + assertStatementInParent("Missing in child statement should be kept new in parent ", 1, stmtNewP_MisC); + } + + @Test + public void testNewParentMissingChild() throws Exception { + RyaStatement stmtNewP_MisC = createRyaStatementUniqueAdd("s_NewP_MisC", "p_NewP_MisC", "o_NewP_MisC", null, parentDao, null); + RyaStatement stmtOldP_OldC = createRyaStatementUniqueAdd("s_OldP_OldC", "p_OldP_OldC", "o_OldP_OldC", LAST_MONTH, parentDao, childDao); + + AccumuloRyaUtils.printTable(PARENT_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, parentConfig); + AccumuloRyaUtils.printTable(CHILD_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, childConfig); + mergeToolRun(YESTERDAY); + assertStatementInParent("Missing in child statement should be kept new in parent ", 1, stmtNewP_MisC); + assertStatementInParent("Statement in parent and child. ", 1, stmtOldP_OldC); + } + + @Test + public void testEmptyParentNewChild() throws Exception { + RyaStatement stmtMisP_NewC_addP_z = createRyaStatementUniqueAdd("zs_MisP_NewC", "zp_MisP_NewC", "zo_MisP_NewC", null , null , childDao); + + AccumuloRyaUtils.printTable(PARENT_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, parentConfig); + AccumuloRyaUtils.printTable(CHILD_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, childConfig); + + mergeToolRun(YESTERDAY); + + AccumuloRyaUtils.printTable(PARENT_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, parentConfig); + AccumuloRyaUtils.printTable(CHILD_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, childConfig); + + assertStatementInParent("Missing Parent New Child should add Parent.", 1, stmtMisP_NewC_addP_z); + + } + + /** + * Test all cases with statements in different splits. + * @throws Exception + */ + @Test + public void testWithDefaultSplits() throws Exception { + addAndVerifySplitableStatements(); + } + + @Test + public void testWithParentSplits() throws Exception { + // set splits, 4 tablets created: <b b*-g g*-v >v + TreeSet<Text> splits = new TreeSet<Text>(); + splits.add(new Text("b")); + splits.add(new Text("g")); + splits.add(new Text("v")); + for (String tableSuffix : AccumuloInstanceDriver.TABLE_NAME_SUFFIXES) { + parentConnector.tableOperations().addSplits(PARENT_TABLE_PREFIX + tableSuffix, splits); + } + addAndVerifySplitableStatements(); + } + + @Test + public void testWithChildSplits() throws Exception { + // set splits, 4 tablets created: <b b*-g g*-v >v + TreeSet<Text> splits = new TreeSet<Text>(); + splits.add(new Text("b")); + splits.add(new Text("g")); + splits.add(new Text("v")); + for (String tableSuffix : AccumuloInstanceDriver.TABLE_NAME_SUFFIXES) { + childConnector.tableOperations().addSplits(CHILD_TABLE_PREFIX + tableSuffix, splits); + } + addAndVerifySplitableStatements(); + } + + @Test + public void testWithParentAndChildSplits() throws Exception { + // set splits, 4 tablets created: <b b*-g g*-v >v + TreeSet<Text> splits = new TreeSet<Text>(); + splits.add(new Text("b")); + splits.add(new Text("g")); + splits.add(new Text("v")); + for (String tableSuffix : AccumuloInstanceDriver.TABLE_NAME_SUFFIXES) { + parentConnector.tableOperations().addSplits(PARENT_TABLE_PREFIX + tableSuffix, splits); + } + for (String tableSuffix : AccumuloInstanceDriver.TABLE_NAME_SUFFIXES) { + childConnector.tableOperations().addSplits(CHILD_TABLE_PREFIX + tableSuffix, splits); + } + addAndVerifySplitableStatements(); + } + + /** + * Not a test, but a setup for all the split tests, all cases. + * Parent | Child | assume that | merge modification + * -------- | ------- | -------------- | ------------------- + * older | missing | child deleted | delete from parent + * newer | missing | parent added | do nothing + * missing | older | parent deleted | do nothing + * missing | newer | child added | add to parent + * + * @throws Exception + */ + private void addAndVerifySplitableStatements() throws Exception { + // Old=older, New=newer, Mis=missing, P=parent, C=child, delP=del from parent, addP=add to parent, Noth=do nothing + RyaStatement stmtOldP_MisC_delP_a = createRyaStatementUniqueAdd("as_OldP_MisC", "ap_OldP_MisC", "ao_OldP_MisC", LAST_MONTH, parentDao, null); + RyaStatement stmtOldP_MisC_delP_f = createRyaStatementUniqueAdd("fs_OldP_MisC", "fp_OldP_MisC", "fo_OldP_MisC", LAST_MONTH, parentDao, null); + RyaStatement stmtOldP_MisC_delP_u = createRyaStatementUniqueAdd("us_OldP_MisC", "up_OldP_MisC", "uo_OldP_MisC", LAST_MONTH, parentDao, null); + RyaStatement stmtOldP_MisC_delP_z = createRyaStatementUniqueAdd("zs_OldP_MisC", "zp_OldP_MisC", "zo_OldP_MisC", LAST_MONTH, parentDao, null); + RyaStatement stmtNewP_MisC_Noth_a = createRyaStatementUniqueAdd("as_NewP_MisC", "ap_NewP_MisC", "ao_NewP_MisC", null , parentDao, null); + RyaStatement stmtNewP_MisC_Noth_f = createRyaStatementUniqueAdd("fs_NewP_MisC", "fp_NewP_MisC", "fo_NewP_MisC", null , parentDao, null); + RyaStatement stmtNewP_MisC_Noth_u = createRyaStatementUniqueAdd("us_NewP_MisC", "up_NewP_MisC", "uo_NewP_MisC", null , parentDao, null); + RyaStatement stmtNewP_MisC_Noth_z = createRyaStatementUniqueAdd("zs_NewP_MisC", "zp_NewP_MisC", "zo_NewP_MisC", null , parentDao, null); + RyaStatement stmtMisP_OldC_Noth_a = createRyaStatementUniqueAdd("as_MisP_OldC", "ap_MisP_OldC", "ao_MisP_OldC", LAST_MONTH, null , childDao); + RyaStatement stmtMisP_OldC_Noth_f = createRyaStatementUniqueAdd("fs_MisP_OldC", "fp_MisP_OldC", "fo_MisP_OldC", LAST_MONTH, null , childDao); + RyaStatement stmtMisP_OldC_Noth_u = createRyaStatementUniqueAdd("us_MisP_OldC", "up_MisP_OldC", "uo_MisP_OldC", LAST_MONTH, null , childDao); + RyaStatement stmtMisP_OldC_addP_z = createRyaStatementUniqueAdd("zs_MisP_OldC", "zp_MisP_OldC", "zo_MisP_OldC", LAST_MONTH, null , childDao); + RyaStatement stmtMisP_NewC_addP_a = createRyaStatementUniqueAdd("as_MisP_NewC", "ap_MisP_NewC", "ao_MisP_NewC", null , null , childDao); + RyaStatement stmtMisP_NewC_addP_f = createRyaStatementUniqueAdd("fs_MisP_NewC", "fp_MisP_NewC", "fo_MisP_NewC", null , null , childDao); + RyaStatement stmtMisP_NewC_addP_u = createRyaStatementUniqueAdd("us_MisP_NewC", "up_MisP_NewC", "uo_MisP_NewC", null , null , childDao); + RyaStatement stmtMisP_NewC_addP_z = createRyaStatementUniqueAdd("zs_MisP_NewC", "zp_MisP_NewC", "zo_MisP_NewC", null , null , childDao); + + AccumuloRyaUtils.printTable(PARENT_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, parentConfig); + AccumuloRyaUtils.printTable(CHILD_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, childConfig); + + mergeToolRun(YESTERDAY); + + AccumuloRyaUtils.printTable(PARENT_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, parentConfig); + AccumuloRyaUtils.printTable(CHILD_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, childConfig); + + String desc = null; + + desc = "Old parent, missing Child, should delete Parent. "; + assertStatementInParent(desc, 0, stmtOldP_MisC_delP_a); + assertStatementInParent(desc, 0, stmtOldP_MisC_delP_f); + assertStatementInParent(desc, 0, stmtOldP_MisC_delP_u); + assertStatementInParent(desc, 0, stmtOldP_MisC_delP_z); + + desc = "New parent, missing Child, should do nothing, leave parent. "; + assertStatementInParent(desc, 1, stmtNewP_MisC_Noth_a); + assertStatementInParent(desc, 1, stmtNewP_MisC_Noth_f); + assertStatementInParent(desc, 1, stmtNewP_MisC_Noth_u); + assertStatementInParent(desc, 1, stmtNewP_MisC_Noth_z); + + desc = "Missing parent, Old Child, should do nothing, missing parent. "; + assertStatementInParent(desc, 0, stmtMisP_OldC_Noth_a); + assertStatementInParent(desc, 0, stmtMisP_OldC_Noth_f); + assertStatementInParent(desc, 0, stmtMisP_OldC_Noth_u); + assertStatementInParent(desc, 0, stmtMisP_OldC_addP_z); + + desc = "Missing parent, New Child, add to parent. "; + assertStatementInParent(desc, 1, stmtMisP_NewC_addP_a); + assertStatementInParent(desc, 1, stmtMisP_NewC_addP_f); + assertStatementInParent(desc, 1, stmtMisP_NewC_addP_u); + assertStatementInParent(desc, 1, stmtMisP_NewC_addP_z); + } + +}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/71b5b1c9/extras/rya.merger/src/test/java/mvm/rya/accumulo/mr/merge/RulesetCopyIT.java ---------------------------------------------------------------------- diff --git a/extras/rya.merger/src/test/java/mvm/rya/accumulo/mr/merge/RulesetCopyIT.java b/extras/rya.merger/src/test/java/mvm/rya/accumulo/mr/merge/RulesetCopyIT.java new file mode 100644 index 0000000..f34b798 --- /dev/null +++ b/extras/rya.merger/src/test/java/mvm/rya/accumulo/mr/merge/RulesetCopyIT.java @@ -0,0 +1,561 @@ +package mvm.rya.accumulo.mr.merge; + +/* + * #%L + * mvm.rya.accumulo.mr.merge + * %% + * Copyright (C) 2014 Rya + * %% + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + * #L% + */ + +import static mvm.rya.accumulo.mr.merge.util.TestUtils.YESTERDAY; +import static mvm.rya.accumulo.mr.merge.util.ToolConfigUtils.makeArgument; + +import java.util.Arrays; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; + +import org.apache.accumulo.core.client.Connector; +import org.apache.hadoop.conf.Configuration; +import org.apache.log4j.Logger; +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; +import org.openrdf.model.Namespace; +import org.openrdf.model.Statement; +import org.openrdf.model.URI; +import org.openrdf.model.impl.URIImpl; +import org.openrdf.model.vocabulary.OWL; +import org.openrdf.model.vocabulary.RDF; +import org.openrdf.model.vocabulary.RDFS; +import org.openrdf.model.vocabulary.XMLSchema; +import org.openrdf.query.BindingSet; +import org.openrdf.query.QueryLanguage; +import org.openrdf.query.QueryResultHandlerException; +import org.openrdf.query.TupleQuery; +import org.openrdf.query.TupleQueryResultHandler; +import org.openrdf.query.TupleQueryResultHandlerException; +import org.openrdf.repository.sail.SailRepository; +import org.openrdf.repository.sail.SailRepositoryConnection; +import org.openrdf.sail.Sail; + +import info.aduna.iteration.CloseableIteration; +import junit.framework.Assert; +import mvm.rya.accumulo.AccumuloRdfConfiguration; +import mvm.rya.accumulo.AccumuloRyaDAO; +import mvm.rya.accumulo.mr.merge.common.InstanceType; +import mvm.rya.accumulo.mr.merge.driver.AccumuloDualInstanceDriver; +import mvm.rya.accumulo.mr.merge.util.AccumuloRyaUtils; +import mvm.rya.accumulo.mr.merge.util.TestUtils; +import mvm.rya.accumulo.mr.utils.MRUtils; +import mvm.rya.api.RdfCloudTripleStoreConfiguration; +import mvm.rya.api.domain.RyaStatement; +import mvm.rya.api.domain.RyaType; +import mvm.rya.api.domain.RyaURI; +import mvm.rya.api.persist.RyaDAOException; +import mvm.rya.api.resolver.RyaToRdfConversions; +import mvm.rya.indexing.accumulo.ConfigUtils; +import mvm.rya.sail.config.RyaSailFactory; + +public class RulesetCopyIT { + private static final Logger log = Logger.getLogger(RulesetCopyIT.class); + + private static final boolean IS_MOCK = false; + private static final String CHILD_SUFFIX = MergeTool.CHILD_SUFFIX; + + private static final String PARENT_PASSWORD = AccumuloDualInstanceDriver.PARENT_PASSWORD; + private static final String PARENT_INSTANCE = AccumuloDualInstanceDriver.PARENT_INSTANCE; + private static final String PARENT_TABLE_PREFIX = AccumuloDualInstanceDriver.PARENT_TABLE_PREFIX; + private static final String PARENT_TOMCAT_URL = "http://rya-example-box:8080";; + + private static final String CHILD_PASSWORD = AccumuloDualInstanceDriver.CHILD_PASSWORD; + private static final String CHILD_INSTANCE = AccumuloDualInstanceDriver.CHILD_INSTANCE; + private static final String CHILD_TABLE_PREFIX = AccumuloDualInstanceDriver.CHILD_TABLE_PREFIX; + private static final String CHILD_TOMCAT_URL = "http://localhost:8080";; + + private static final String QUERY_PREFIXES; + + private static AccumuloRdfConfiguration parentConfig; + private static AccumuloRdfConfiguration childConfig; + + private static AccumuloDualInstanceDriver accumuloDualInstanceDriver; + private static CopyTool rulesetTool = null; + private static AccumuloRyaDAO parentDao; + + private static Map<String, String> prefixes = new HashMap<>(); + static { + prefixes.put("test:", "http://example.com#";); + prefixes.put("alt:", "http://example.com/alternate#";); + prefixes.put("time:", "http://www.w3.org/2006/time#";); + prefixes.put("geo:", "http://www.opengis.net/ont/geosparql#";); + prefixes.put("fts:", "http://rdf.useekm.com/fts#";); + prefixes.put("tempo:", "tag:rya-rdf.org,2015:temporal#"); + prefixes.put("rdf:", RDF.NAMESPACE); + prefixes.put("rdfs:", RDFS.NAMESPACE); + prefixes.put("owl:", OWL.NAMESPACE); + StringBuilder sb = new StringBuilder(); + for (String prefix : prefixes.keySet()) { + sb.append("PREFIX " + prefix + " <" + prefixes.get(prefix) + ">\n"); + } + QUERY_PREFIXES = sb.toString(); + } + + private static RyaURI substitute(String uri) { + for (String prefix : prefixes.keySet()) { + if (uri.startsWith(prefix)) { + return new RyaURI(uri.replace(prefix, prefixes.get(prefix))); + } + } + return new RyaURI(uri); + } + + private static RyaStatement statement(String s, String p, RyaType o) { + RyaStatement ryaStatement = new RyaStatement(substitute(s), substitute(p), o); + ryaStatement.setTimestamp(YESTERDAY.getTime()); + return ryaStatement; + } + + private static RyaStatement statement(String s, String p, String o) { + return statement(s, p, substitute(o)); + } + + private static RyaType literal(String lit) { + return new RyaType(lit); + } + + private static RyaType literal(String lit, URI type) { + return new RyaType(type, lit); + } + + @BeforeClass + public static void setUpPerClass() throws Exception { + accumuloDualInstanceDriver = new AccumuloDualInstanceDriver(IS_MOCK, true, true, false, false); + accumuloDualInstanceDriver.setUpInstances(); + } + + @Before + public void setUpPerTest() throws Exception { + parentConfig = accumuloDualInstanceDriver.getParentConfig(); + childConfig = accumuloDualInstanceDriver.getChildConfig(); + accumuloDualInstanceDriver.setUpTables(); + accumuloDualInstanceDriver.setUpConfigs(); + accumuloDualInstanceDriver.setUpDaos(); + parentDao = accumuloDualInstanceDriver.getParentDao(); + } + + @After + public void tearDownPerTest() throws Exception { + log.info("tearDownPerTest(): tearing down now."); + accumuloDualInstanceDriver.tearDownDaos(); + accumuloDualInstanceDriver.tearDownTables(); + } + + @AfterClass + public static void tearDownPerClass() throws Exception { + if (rulesetTool != null) { + rulesetTool.shutdown(); + } + accumuloDualInstanceDriver.tearDown(); + } + + private AccumuloRyaDAO runRulesetCopyTest(RyaStatement[] solutionStatements, RyaStatement[] copyStatements, + RyaStatement[] irrelevantStatements, String query, int numSolutions, boolean infer) throws Exception { + log.info("Adding data to parent..."); + parentDao.add(Arrays.asList(solutionStatements).iterator()); + parentDao.add(Arrays.asList(copyStatements).iterator()); + parentDao.add(Arrays.asList(irrelevantStatements).iterator()); + + log.info("Copying from parent tables:"); + for (String table : accumuloDualInstanceDriver.getParentTableList()) { + AccumuloRyaUtils.printTablePretty(table, parentConfig, false); + } + + rulesetTool = new CopyTool(); + rulesetTool.setupAndRun(new String[] { + makeArgument(MRUtils.AC_MOCK_PROP, Boolean.toString(IS_MOCK)), + makeArgument(MRUtils.AC_INSTANCE_PROP, PARENT_INSTANCE), + makeArgument(MRUtils.AC_USERNAME_PROP, accumuloDualInstanceDriver.getParentUser()), + makeArgument(MRUtils.AC_PWD_PROP, PARENT_PASSWORD), + makeArgument(MRUtils.TABLE_PREFIX_PROPERTY, PARENT_TABLE_PREFIX), + makeArgument(MRUtils.AC_AUTH_PROP, accumuloDualInstanceDriver.getParentAuth()), + makeArgument(MRUtils.AC_ZK_PROP, accumuloDualInstanceDriver.getParentZooKeepers()), + makeArgument(CopyTool.PARENT_TOMCAT_URL_PROP, PARENT_TOMCAT_URL), + makeArgument(MRUtils.AC_MOCK_PROP + CHILD_SUFFIX, Boolean.toString(IS_MOCK)), + makeArgument(MRUtils.AC_INSTANCE_PROP + CHILD_SUFFIX, CHILD_INSTANCE), + makeArgument(MRUtils.AC_USERNAME_PROP + CHILD_SUFFIX, accumuloDualInstanceDriver.getChildUser()), + makeArgument(MRUtils.AC_PWD_PROP + CHILD_SUFFIX, CHILD_PASSWORD), + makeArgument(MRUtils.TABLE_PREFIX_PROPERTY + CHILD_SUFFIX, CHILD_TABLE_PREFIX), + makeArgument(MRUtils.AC_AUTH_PROP + CHILD_SUFFIX, accumuloDualInstanceDriver.getChildAuth() != null ? accumuloDualInstanceDriver.getChildAuth() : null), + makeArgument(MRUtils.AC_ZK_PROP + CHILD_SUFFIX, accumuloDualInstanceDriver.getChildZooKeepers() != null ? accumuloDualInstanceDriver.getChildZooKeepers() : "localhost"), + makeArgument(CopyTool.CHILD_TOMCAT_URL_PROP, CHILD_TOMCAT_URL), + makeArgument(CopyTool.CREATE_CHILD_INSTANCE_TYPE_PROP, (IS_MOCK ? InstanceType.MOCK : InstanceType.MINI).toString()), + makeArgument(CopyTool.QUERY_STRING_PROP, query), + makeArgument(CopyTool.USE_COPY_QUERY_SPARQL, "true"), + makeArgument(RdfCloudTripleStoreConfiguration.CONF_INFER, Boolean.toString(infer)) + }); + + Configuration toolConfig = rulesetTool.getConf(); + childConfig.set(MRUtils.AC_ZK_PROP, toolConfig.get(MRUtils.AC_ZK_PROP + CHILD_SUFFIX)); + MergeTool.setDuplicateKeys(childConfig); + + log.info("Finished running tool."); + + // Child instance has now been created + Connector childConnector = ConfigUtils.getConnector(childConfig); + accumuloDualInstanceDriver.getChildAccumuloInstanceDriver().setConnector(childConnector); + accumuloDualInstanceDriver.getChildAccumuloInstanceDriver().setUpTables(); + accumuloDualInstanceDriver.getChildAccumuloInstanceDriver().setUpDao(); + AccumuloRyaDAO childDao = accumuloDualInstanceDriver.getChildDao(); + + log.info("Resulting child tables:"); + for (String table : accumuloDualInstanceDriver.getChildTableList()) { + AccumuloRyaUtils.printTablePretty(table, childConfig, false); + } + + for (RyaStatement solution : solutionStatements) { + Statement stmt = RyaToRdfConversions.convertStatement(solution); + TestUtils.assertStatementInInstance("Child missing solution statement " + stmt, + 1, solution, childDao, childConfig); + } + for (RyaStatement copied : copyStatements) { + Statement stmt = RyaToRdfConversions.convertStatement(copied); + TestUtils.assertStatementInInstance("Child missing relevant statement " + stmt, + 1, copied, childDao, childConfig); + } + for (RyaStatement irrelevant : irrelevantStatements) { + Statement stmt = RyaToRdfConversions.convertStatement(irrelevant); + TestUtils.assertStatementInInstance("Should not have copied irrelevant statement " + stmt, + 0, irrelevant, childDao, childConfig); + } + + Set<BindingSet> parentSolutions = runQuery(query, parentConfig); + if (parentSolutions.isEmpty()) { + log.info("No solutions to query in parent"); + } + else { + for (BindingSet bs : parentSolutions) { + log.info("Parent yields query solution: " + bs); + } + } + Set<BindingSet> childSolutions = runQuery(query, childConfig); + if (childSolutions.isEmpty()) { + log.info("No solutions to query in child"); + } + else { + for (BindingSet bs : childSolutions) { + log.info("Child yields query solution: " + bs); + } + } + Assert.assertEquals("Query results should be the same:", parentSolutions, childSolutions); + Assert.assertEquals("Incorrect number of solutions:", numSolutions, childSolutions.size()); + return childDao; + } + + private Set<BindingSet> runQuery(String query, Configuration conf) throws Exception { + SailRepository repository = null; + SailRepositoryConnection conn = null; + try { + Sail extSail = RyaSailFactory.getInstance(conf); + repository = new SailRepository(extSail); + repository.initialize(); + conn = repository.getConnection(); + ResultHandler handler = new ResultHandler(); + TupleQuery tq = conn.prepareTupleQuery(QueryLanguage.SPARQL, query); + tq.evaluate(handler); + return handler.getSolutions(); + } + finally { + if (conn != null) { + conn.close(); + } + if (repository != null) { + repository.shutDown(); + } + } + } + + private static class ResultHandler implements TupleQueryResultHandler { + private Set<BindingSet> solutions = new HashSet<>(); + public Set<BindingSet> getSolutions() { + return solutions; + } + @Override + public void startQueryResult(List<String> arg0) throws TupleQueryResultHandlerException { + } + @Override + public void handleSolution(BindingSet arg0) throws TupleQueryResultHandlerException { + solutions.add(arg0); + } + @Override + public void endQueryResult() throws TupleQueryResultHandlerException { + } + @Override + public void handleBoolean(boolean arg0) throws QueryResultHandlerException { + } + @Override + public void handleLinks(List<String> arg0) throws QueryResultHandlerException { + } + } + + @Test + public void testRulesetCopyTool() throws Exception { + // Should be copied and are involved in the solution: + RyaStatement[] solutionStatements = { + statement("test:FullProfessor1", "rdf:type", "test:FullProfessor"), + statement("test:GraduateStudent1", "test:advisor", "test:FullProfessor1"), + statement("test:FullProfessor1", "test:telephone", literal("123-456-7890")), + statement("test:University1", "test:telephone", literal("555")), + statement("test:FullProfessor1", "test:worksFor", "test:University1"), + statement("test:FullProfessor1", "test:hired", literal("2001-01-01T04:01:02.000Z", XMLSchema.DATETIME)), + statement("test:University1", "geo:asWKT", literal("Point(-77.03524 38.889468)", new URIImpl("http://www.opengis.net/ont/geosparql#wktLiteral";))) + }; + // These aren't solutions but should be copied: + RyaStatement[] copyStatements = { + statement("test:FullProfessor2", "rdf:type", "test:FullProfessor"), + statement("test:GraduateStudent1", "test:advisor", "test:AssistantProfessor1"), + statement("test:GraduateStudent1", "test:telephone", literal("555-123-4567")), + statement("test:FullProfessor1", "test:telephone", literal("567-8901")), + statement("test:University1", "test:telephone", literal("800-123-4567")) + }; + // Should not be copied: + RyaStatement[] irrelevantStatements = { + statement("test:GraduateStudent2", "test:advisor", "test:FullProfessor1"), + statement("test:UndergraduateStudent1", "rdf:type", "test:UndergraduateStudent"), + statement("test:UndergraduateStudent2", "rdf:type", "test:UndergraduateStudent"), + statement("test:GraduateStudent1", "rdf:type", "test:GraduateStudent"), + statement("test:GraduateStudent1", "test:name", literal("GraduateStudent1")), + statement("test:UndergraduateStudent2", "test:name", literal("UndergraduateStudent2")), + statement("test:Course1", "test:name", literal("Course1")), + statement("test:GraduateStudent1", "test:emailAddress", literal("graduatestude...@department0.university0.edu")), + statement("test:GraduateStudent1", "test:teachingAssistantOf", "test:Course1"), + statement("test:GraduateStudent2", "test:undergraduateDegreeFrom", "test:University1"), + statement("test:AssistantProfessor1", "test:teacherOf", "test:Course1"), + statement("test:FullProfessor1", "test:telephone", literal("xxx-xxx-xxxx")), + statement("test:University1", "test:telephone", literal("0000")), + // If inferencing is disabled, these shouldn't matter: + statement("test:employs", "owl:inverseOf", "test:worksFor"), + statement("test:University1", "test:employs", "test:FullProfessor2") + }; + + String query = QUERY_PREFIXES + "SELECT * {\n" + + " test:GraduateStudent1 test:advisor ?person .\n" + + " ?person rdf:type test:FullProfessor .\n" + + " ?person test:telephone ?number .\n" + + " ?person test:worksFor ?university .\n" + + " FILTER regex(?number, \"...-...-....\")\n" + + " ?university geo:asWKT ?place .\n" + + " FILTER regex(?number, \"^[0-9]\")\n" + + " ?university test:telephone ?universityNumber\n" + + " FILTER regex(?universityNumber, \"^5\")\n" +// Would be good to test index functions, but indexing is unreliable (see RYA-72): + + " ?person test:hired ?time .\n" +// + " FILTER(tempo:after(?time, '2000-01-01T01:01:03-08:00'))\n" + + "}"; + + int parentNamespaceCount = 2; + int childNamespaceCount = 0; + parentDao.addNamespace("ns1", "http://www.example.com/ns1#";); + parentDao.addNamespace("ns2", "http://www.example.com/ns2#";); + // Run the test + AccumuloRyaDAO childDao = runRulesetCopyTest(solutionStatements, copyStatements, irrelevantStatements, query, 1, false); + // Verify namespaces were copied + CloseableIteration<Namespace, RyaDAOException> nsIter = childDao.iterateNamespace(); + while (nsIter.hasNext()) { + childNamespaceCount++; + nsIter.next(); + } + Assert.assertEquals("Incorrect number of namespaces copied to child:", parentNamespaceCount, childNamespaceCount); + log.info("DONE"); + } + + /* + * Tests subclass and subproperty inference. Based on standard LUBM query #5. + */ + @Test + public void testRulesetCopyHierarchy() throws Exception { + RyaStatement[] solutionStatements = { + statement("test:p1", "rdf:type", "test:Professor"), + statement("test:p1", "test:worksFor", "test:Department0"), + statement("test:p2", "rdf:type", "test:FullProfessor"), + statement("test:p2", "test:headOf", "test:Department0"), + }; + RyaStatement[] copyStatements = { + // schema: + statement("test:Professor", "rdfs:subClassOf", "test:Person"), + statement("test:Student", "rdfs:subClassOf", "test:Person"), + statement("test:GraduateStudent", "rdfs:subClassOf", "test:Student"), + statement("test:FullProfessor", "rdfs:subClassOf", "test:Professor"), + statement("test:AssistantProfessor", "rdfs:subClassOf", "test:Professor"), + statement("test:worksFor", "rdfs:subPropertyOf", "test:memberOf"), + statement("test:headOf", "rdfs:subPropertyOf", "test:worksFor"), + // data: + statement("test:s1", "rdf:type", "test:Student"), + statement("test:ap1", "rdf:type", "test:AssistantProfessor"), + statement("test:gs1", "rdf:type", "test:GraduateStudent"), + }; + RyaStatement[] otherStatements = { + // schema: + statement("test:worksFor", "rdfs:subPropertyOf", "test:affiliatedWith"), + statement("test:Person", "rdfs:subClassOf", "test:Animal"), + // data: + statement("test:University0", "test:hasSubOrganizationOf", "test:Department0"), + statement("test:a1", "rdf:type", "test:Animal") + }; + String query = QUERY_PREFIXES + "SELECT * {\n" + + " ?X rdf:type test:Person .\n" + + " ?X test:memberOf test:Department0 .\n" + + "}"; + runRulesetCopyTest(solutionStatements, copyStatements, otherStatements, query, 2, true); + log.info("DONE"); + } + + /** + * Test the ruleset copy with owl:sameAs inference + */ + @Test + public void testRulesetCopySameAs() throws Exception { + String query = QUERY_PREFIXES + "SELECT * {\n" + + " {\n" + + " ?X test:worksFor test:Department0 .\n" + + " ?X rdf:type test:Student\n" + + " } UNION {\n" + + " test:p1 rdf:type test:Professor .\n" + + " test:p1 test:worksFor test:CSDept .\n" + + " }\n" + + "}"; + RyaStatement[] solutionStatements = { + statement("test:s1", "test:worksFor", "test:CSDept"), + statement("test:s1", "rdf:type", "test:Student"), + statement("test:p1", "rdf:type", "test:Professor"), + statement("alt:p1", "test:worksFor", "test:CSDept"), + statement("test:CSDept", "owl:sameAs", "test:Department0"), + statement("test:p1", "owl:sameAs", "alt:p1"), + statement("test:JohnDoe", "owl:sameAs", "alt:p1") + }; + RyaStatement[] copyStatements = { + statement("test:s2", "rdf:type", "test:Student"), + statement("alt:s2", "test:worksFor", "test:CSDept"), + statement("test:s3", "test:worksFor", "test:CSDept") + }; + RyaStatement[] otherStatements = { + // sameAs inference only expands constants: + statement("test:s2", "owl:sameAs", "alt:s2"), + // sameAs inference not applied to rdf:type statements: + statement("alt:p1", "rdf:type", "test:Professor"), + statement("test:s3", "rdf:type", "alt:Student"), + statement("test:Student", "owl:sameAs", "alt:Student") + }; + runRulesetCopyTest(solutionStatements, copyStatements, otherStatements, query, 2, true); + log.info("DONE"); + } + + /** + * Test the ruleset copy with owl:TransitiveProperty inference. + * Based on LUBM test query #11. + */ + @Test + public void testRulesetCopyTransitive() throws Exception { + String query = QUERY_PREFIXES + "SELECT * {\n" + // Note: we get spurious results if the order of these are switched (see RYA-71): + + " ?X test:subOrganizationOf test:University0 .\n" + + " ?X rdf:type test:ResearchGroup .\n" + + "}"; + RyaStatement[] solutionStatements = { + statement("test:subOrganizationOf", "rdf:type", "owl:TransitiveProperty"), + statement("test:ResearchGroup0", "rdf:type", "test:ResearchGroup"), + statement("test:ResearchGroup0", "test:subOrganizationOf", "test:Department0"), + statement("test:Department0", "test:subOrganizationOf", "test:University0"), + statement("test:Subgroup0", "rdf:type", "test:ResearchGroup"), + statement("test:Subgroup0", "test:subOrganizationOf", "test:ResearchGroup0") + }; + RyaStatement[] copyStatements = { + statement("test:ResearchGroupA", "rdf:type", "test:ResearchGroup"), + statement("test:ResearchGroupA", "test:subOrganizationOf", "test:DepartmentA"), + statement("test:DepartmentA", "test:subOrganizationOf", "test:UniversityA"), + statement("test:OtherGroup0", "test:subOrganizationOf", "test:Department0"), + statement("test:Department1", "test:subOrganizationOf", "test:University0") + }; + RyaStatement[] otherStatements = { + statement("test:University0", "rdf:type", "test:University"), + statement("test:Department0", "test:affiliatedWith", "test:University0") + }; + runRulesetCopyTest(solutionStatements, copyStatements, otherStatements, query, 2, true); + log.info("DONE"); + } + + /** + * Test the ruleset copy with owl:inverseOf and owl:subPropertyOf inference. + * Based on LUBM test query #13. + */ + @Test + public void testRulesetCopyInverse() throws Exception { + String query = QUERY_PREFIXES + "SELECT * {\n" + + " ?X rdf:type test:Person .\n" + + " test:University0 test:hasAlumnus ?X .\n" + + "}"; + RyaStatement[] solutionStatements = { + statement("test:s1", "rdf:type", "test:Person"), + statement("test:p1", "rdf:type", "test:Person"), + statement("test:s1", "test:undergraduateDegreeFrom", "test:University0"), + statement("test:p1", "test:doctoralDegreeFrom", "test:University0"), + statement("test:undergraduateDegreeFrom", "rdfs:subPropertyOf", "test:degreeFrom"), + statement("test:doctoralDegreeFrom", "rdfs:subPropertyOf", "test:degreeFrom"), + statement("test:hasAlumnus", "owl:inverseOf", "test:degreeFrom") + }; + RyaStatement[] copyStatements = { + statement("test:mastersDegreeFrom", "rdfs:subPropertyOf", "test:degreeFrom"), + statement("test:s2", "test:mastersDegreeFrom", "test:University0"), + }; + RyaStatement[] otherStatements = { + statement("test:p1", "test:mastersDegreeFrom", "test:University1"), + }; + runRulesetCopyTest(solutionStatements, copyStatements, otherStatements, query, 2, true); + log.info("DONE"); + } + + /** + * Test the ruleset copy with owl:SymmetricProperty inference, combined with rdfs:subPropertyOf. + */ + @Test + public void testRulesetCopySymmetry() throws Exception { + String query = QUERY_PREFIXES + "SELECT * {\n" + + " ?X rdf:type test:Person .\n" + + " ?X test:knows test:Alice .\n" + + "}"; + RyaStatement[] solutionStatements = { + statement("test:Alice", "test:knows", "test:Bob"), + statement("test:Alice", "test:friendsWith", "test:Carol"), + statement("test:Bob", "rdf:type", "test:Person"), + statement("test:Carol", "rdf:type", "test:Person"), + statement("test:friendsWith", "rdfs:subPropertyOf", "test:knows"), + statement("test:knows", "rdf:type", "owl:SymmetricProperty") + }; + RyaStatement[] copyStatements = { + statement("test:Alice", "rdf:type", "test:Person"), + statement("test:Eve", "rdf:type", "test:Person") + }; + RyaStatement[] otherStatements = { + statement("test:Carol", "test:knows", "test:Eve"), + statement("test:Bob", "test:friendsWith", "test:Carol") + }; + runRulesetCopyTest(solutionStatements, copyStatements, otherStatements, query, 2, true); + log.info("DONE"); + } +} http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/71b5b1c9/extras/rya.merger/src/test/java/mvm/rya/accumulo/mr/merge/demo/CopyToolDemo.java ---------------------------------------------------------------------- diff --git a/extras/rya.merger/src/test/java/mvm/rya/accumulo/mr/merge/demo/CopyToolDemo.java b/extras/rya.merger/src/test/java/mvm/rya/accumulo/mr/merge/demo/CopyToolDemo.java new file mode 100644 index 0000000..2ab2b90 --- /dev/null +++ b/extras/rya.merger/src/test/java/mvm/rya/accumulo/mr/merge/demo/CopyToolDemo.java @@ -0,0 +1,294 @@ +package mvm.rya.accumulo.mr.merge.demo; + +import static mvm.rya.accumulo.mr.merge.util.TestUtils.LAST_MONTH; +import static mvm.rya.accumulo.mr.merge.util.TestUtils.TODAY; +import static mvm.rya.accumulo.mr.merge.util.TestUtils.YESTERDAY; +import static mvm.rya.accumulo.mr.merge.util.TestUtils.createRyaStatement; +import static mvm.rya.accumulo.mr.merge.util.ToolConfigUtils.makeArgument; + +import java.util.Date; +import java.util.Iterator; +import java.util.List; +import java.util.Map.Entry; +import java.util.Random; +import java.util.SortedSet; +import java.util.TreeSet; + +import org.apache.accumulo.core.client.AccumuloException; +import org.apache.accumulo.core.client.AccumuloSecurityException; +import org.apache.accumulo.core.client.Connector; +import org.apache.accumulo.core.client.Scanner; +import org.apache.accumulo.core.client.admin.SecurityOperations; +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.Value; +import org.apache.accumulo.core.file.rfile.bcfile.Compression.Algorithm; +import org.apache.accumulo.core.security.Authorizations; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.io.Text; +import org.apache.log4j.Logger; + +import mvm.rya.accumulo.AccumuloRdfConfiguration; +import mvm.rya.accumulo.AccumuloRyaDAO; +import mvm.rya.accumulo.mr.merge.CopyTool; +import mvm.rya.accumulo.mr.merge.MergeTool; +import mvm.rya.accumulo.mr.merge.common.InstanceType; +import mvm.rya.accumulo.mr.merge.demo.util.DemoUtilities; +import mvm.rya.accumulo.mr.merge.demo.util.DemoUtilities.LoggingDetail; +import mvm.rya.accumulo.mr.merge.driver.AccumuloDualInstanceDriver; +import mvm.rya.accumulo.mr.merge.util.AccumuloRyaUtils; +import mvm.rya.accumulo.mr.merge.util.TimeUtils; +import mvm.rya.accumulo.mr.utils.MRUtils; +import mvm.rya.api.RdfCloudTripleStoreConstants; +import mvm.rya.api.domain.RyaStatement; +import mvm.rya.indexing.accumulo.ConfigUtils; + +/** + * Tests for {@link CopyTool}. + */ +public class CopyToolDemo { + private static final Logger log = Logger.getLogger(CopyToolDemo.class); + + private static final boolean IS_MOCK = true; + private static final boolean USE_TIME_SYNC = false; + private static final boolean USE_COPY_FILE_OUTPUT = false; + private static final boolean USE_COPY_FILE_IMPORT = false; + private static final boolean IS_PROMPTING_ENABLED = true; + private static final boolean IS_START_TIME_DIALOG_ENABLED = true; + private static final LoggingDetail LOGGING_DETAIL = LoggingDetail.LIGHT; + + private static final String CHILD_SUFFIX = MergeTool.CHILD_SUFFIX; + + private static final String PARENT_PASSWORD = AccumuloDualInstanceDriver.PARENT_PASSWORD; + private static final String PARENT_INSTANCE = AccumuloDualInstanceDriver.PARENT_INSTANCE; + private static final String PARENT_TABLE_PREFIX = AccumuloDualInstanceDriver.PARENT_TABLE_PREFIX; + private static final String PARENT_AUTH = AccumuloDualInstanceDriver.PARENT_AUTH; + private static final String PARENT_TOMCAT_URL = "http://rya-example-box:8080";; + + private static final String CHILD_PASSWORD = AccumuloDualInstanceDriver.CHILD_PASSWORD; + private static final String CHILD_INSTANCE = AccumuloDualInstanceDriver.CHILD_INSTANCE; + private static final String CHILD_TABLE_PREFIX = AccumuloDualInstanceDriver.CHILD_TABLE_PREFIX; + private static final String CHILD_TOMCAT_URL = "http://localhost:8080";; + + private AccumuloRyaDAO parentDao; + + private AccumuloRdfConfiguration parentConfig; + private AccumuloRdfConfiguration childConfig; + + private AccumuloDualInstanceDriver accumuloDualInstanceDriver; + private CopyTool copyTool = null; + + public static void main(String args[]) { + DemoUtilities.setupLogging(LOGGING_DETAIL); + log.info("Setting up Copy Tool Demo"); + + Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { + @Override + public void uncaughtException(Thread thread, Throwable throwable) { + log.fatal("Uncaught exception in " + thread.getName(), throwable); + } + }); + + final CopyToolDemo copyToolDemo = new CopyToolDemo(); + Runtime.getRuntime().addShutdownHook(new Thread() { + @Override + public void run() { + log.info("Shutting down..."); + try { + copyToolDemo.tearDown(); + } catch (Exception e) { + log.error("Error while shutting down", e); + } finally { + log.info("Done shutting down"); + } + } + }); + + try { + copyToolDemo.setUp(); + copyToolDemo.testCopyTool(); + } catch (Exception e) { + log.error("Error while testing copy tool", e); + } finally { + try { + copyToolDemo.tearDown(); + } catch (Exception e) { + log.error("Error shutting down copy tool", e); + } + } + + System.exit(0); + } + + public void setUp() throws Exception { + accumuloDualInstanceDriver = new AccumuloDualInstanceDriver(IS_MOCK, true, true, false, false); + accumuloDualInstanceDriver.setUpInstances(); + + accumuloDualInstanceDriver.setUpTables(); + + accumuloDualInstanceDriver.setUpDaos(); + + accumuloDualInstanceDriver.setUpConfigs(); + + parentConfig = accumuloDualInstanceDriver.getParentConfig(); + childConfig = accumuloDualInstanceDriver.getChildConfig(); + parentDao = accumuloDualInstanceDriver.getParentDao(); + } + + public void tearDown() throws Exception { + log.info("Tearing down..."); + accumuloDualInstanceDriver.tearDown(); + if (copyTool != null) { + copyTool.shutdown(); + } + } + + private void copyToolRun(Date startDate) throws AccumuloException, AccumuloSecurityException { + copyTool = new CopyTool(); + copyTool.setupAndRun(new String[] { + makeArgument(MRUtils.AC_MOCK_PROP, Boolean.toString(IS_MOCK)), + makeArgument(MRUtils.AC_INSTANCE_PROP, PARENT_INSTANCE), + makeArgument(MRUtils.AC_USERNAME_PROP, accumuloDualInstanceDriver.getParentUser()), + makeArgument(MRUtils.AC_PWD_PROP, PARENT_PASSWORD), + makeArgument(MRUtils.TABLE_PREFIX_PROPERTY, PARENT_TABLE_PREFIX), + makeArgument(MRUtils.AC_AUTH_PROP, accumuloDualInstanceDriver.getParentAuths().toString()), + makeArgument(MRUtils.AC_ZK_PROP, accumuloDualInstanceDriver.getParentZooKeepers()), + makeArgument(CopyTool.PARENT_TOMCAT_URL_PROP, PARENT_TOMCAT_URL), + makeArgument(MRUtils.AC_MOCK_PROP + CHILD_SUFFIX, Boolean.toString(IS_MOCK)), + makeArgument(MRUtils.AC_INSTANCE_PROP + CHILD_SUFFIX, CHILD_INSTANCE), + makeArgument(MRUtils.AC_USERNAME_PROP + CHILD_SUFFIX, accumuloDualInstanceDriver.getChildUser()), + makeArgument(MRUtils.AC_PWD_PROP + CHILD_SUFFIX, CHILD_PASSWORD), + makeArgument(MRUtils.TABLE_PREFIX_PROPERTY + CHILD_SUFFIX, CHILD_TABLE_PREFIX), + makeArgument(MRUtils.AC_AUTH_PROP + CHILD_SUFFIX, accumuloDualInstanceDriver.getChildAuths() != null ? accumuloDualInstanceDriver.getChildAuths().toString() : null), + makeArgument(MRUtils.AC_ZK_PROP + CHILD_SUFFIX, accumuloDualInstanceDriver.getChildZooKeepers() != null ? accumuloDualInstanceDriver.getChildZooKeepers() : "localhost"), + makeArgument(CopyTool.CHILD_TOMCAT_URL_PROP, CHILD_TOMCAT_URL), + makeArgument(CopyTool.COPY_TABLE_LIST_PROP, !USE_COPY_FILE_IMPORT ? PARENT_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX : ""), + makeArgument(CopyTool.CREATE_CHILD_INSTANCE_TYPE_PROP, (IS_MOCK ? InstanceType.MOCK : InstanceType.MINI).toString()), + makeArgument(CopyTool.NTP_SERVER_HOST_PROP, TimeUtils.DEFAULT_TIME_SERVER_HOST), + makeArgument(CopyTool.USE_NTP_SERVER_PROP, Boolean.toString(USE_TIME_SYNC)), + makeArgument(CopyTool.USE_COPY_FILE_OUTPUT, Boolean.toString(USE_COPY_FILE_OUTPUT)), + makeArgument(CopyTool.COPY_FILE_OUTPUT_PATH, "/test/copy_tool_file_output/"), + makeArgument(CopyTool.COPY_FILE_OUTPUT_COMPRESSION_TYPE, Algorithm.GZ.getName()), + makeArgument(CopyTool.USE_COPY_FILE_OUTPUT_DIRECTORY_CLEAR, Boolean.toString(true)), + makeArgument(CopyTool.COPY_FILE_IMPORT_DIRECTORY, "resources/test/copy_tool_file_output/"), + makeArgument(CopyTool.USE_COPY_FILE_IMPORT, Boolean.toString(USE_COPY_FILE_IMPORT)), + //makeArgument(CopyTool.COPY_TABLE_LIST_PROP, Joiner.on(",").join(accumuloDualInstanceDriver.getParentTableList())), + makeArgument(MergeTool.START_TIME_PROP, MergeTool.getStartTimeString(startDate, IS_START_TIME_DIALOG_ENABLED)) + }); + + Configuration toolConfig = copyTool.getConf(); + String zooKeepers = toolConfig.get(MRUtils.AC_ZK_PROP + CHILD_SUFFIX); + MergeTool.setDuplicateKeysForProperty(childConfig, MRUtils.AC_ZK_PROP, zooKeepers); + + if (USE_COPY_FILE_OUTPUT) { + // Set up the child tables now to test importing the files back into the child instance + String childTableName = CHILD_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX; + try { + copyTool.createTableIfNeeded(childTableName); + copyTool.importFilesToChildTable(childTableName); + } catch (Exception e) { + log.error("Failed to import files into child instance.", e); + } + } + + log.info("Finished running tool."); + } + + public void testCopyTool() throws Exception { + log.info(""); + log.info("Setting up initial state of parent before copying to child..."); + log.info("Adding data to parent..."); + + int numRowsNotToCopy = 80; + int numRowsToCopy = 20; + + // Create Rya Statement before last month which won't be copied + Random random = new Random(); + + for (int i = 1; i <= numRowsNotToCopy; i++) { + long randTimeBeforeLastMonth = DemoUtilities.randLong(0, LAST_MONTH.getTime()); + String randVis = random.nextBoolean() ? PARENT_AUTH : ""; + RyaStatement ryaStatementOutOfTimeRange = createRyaStatement("Nobody", "sees", "me " + i, new Date(randTimeBeforeLastMonth)); + ryaStatementOutOfTimeRange.setColumnVisibility(randVis.getBytes()); + parentDao.add(ryaStatementOutOfTimeRange); + } + + for (int i = 1; i <= numRowsToCopy; i++) { + long randTimeAfterYesterdayAndBeforeToday = DemoUtilities.randLong(YESTERDAY.getTime(), TODAY.getTime()); + String randVis = random.nextBoolean() ? PARENT_AUTH : ""; + RyaStatement ryaStatementShouldCopy = createRyaStatement("bob", "copies", "susan " + i, new Date(randTimeAfterYesterdayAndBeforeToday)); + ryaStatementShouldCopy.setColumnVisibility(randVis.getBytes()); + parentDao.add(ryaStatementShouldCopy); + } + + if (USE_COPY_FILE_OUTPUT) { + // Set up table splits + SortedSet<Text> splits = new TreeSet<>(); + for (char alphabet = 'a'; alphabet <= 'e'; alphabet++) { + Text letter = new Text(alphabet + ""); + splits.add(letter); + } + parentDao.getConnector().tableOperations().addSplits(PARENT_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, splits); + } + + log.info("Added " + (numRowsNotToCopy + numRowsToCopy) + " rows to parent SPO table."); + log.info("Parent SPO table output below:"); + DemoUtilities.promptEnterKey(IS_PROMPTING_ENABLED); + + + AccumuloRyaUtils.printTablePretty(PARENT_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, parentConfig); + //AccumuloRyaUtils.printTablePretty(CHILD_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, childConfig); + + log.info(""); + log.info("Total Rows in table: " + (numRowsNotToCopy + numRowsToCopy)); + log.info("Number of Rows NOT to copy (out of time range): " + numRowsNotToCopy); + log.info("Number of Rows to copy (in time range): " + numRowsToCopy); + log.info(""); + + DemoUtilities.promptEnterKey(IS_PROMPTING_ENABLED); + + log.info("Starting copy tool. Copying all data after the specified start time: " + YESTERDAY); + log.info(""); + + copyToolRun(YESTERDAY); + + + // Copy Tool made child instance so hook the tables and dao into the driver. + String childUser = accumuloDualInstanceDriver.getChildUser(); + Connector childConnector = ConfigUtils.getConnector(childConfig); + accumuloDualInstanceDriver.getChildAccumuloInstanceDriver().setConnector(childConnector); + + accumuloDualInstanceDriver.getChildAccumuloInstanceDriver().setUpTables(); + + accumuloDualInstanceDriver.getChildAccumuloInstanceDriver().setUpDao(); + + + // Update child config to include changes made from copy process + SecurityOperations childSecOps = accumuloDualInstanceDriver.getChildSecOps(); + Authorizations newChildAuths = AccumuloRyaUtils.addUserAuths(childUser, childSecOps, PARENT_AUTH); + childSecOps.changeUserAuthorizations(childUser, newChildAuths); + String childAuthString = newChildAuths.toString(); + List<String> duplicateKeys = MergeTool.DUPLICATE_KEY_MAP.get(MRUtils.AC_AUTH_PROP); + childConfig.set(MRUtils.AC_AUTH_PROP, childAuthString); + for (String key : duplicateKeys) { + childConfig.set(key, childAuthString); + } + + + //AccumuloRyaUtils.printTablePretty(CHILD_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_PO_SUFFIX, childConfig); + //AccumuloRyaUtils.printTablePretty(CHILD_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_OSP_SUFFIX, childConfig); + AccumuloRyaUtils.printTablePretty(CHILD_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, childConfig); + + Scanner scanner = AccumuloRyaUtils.getScanner(CHILD_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, childConfig); + Iterator<Entry<Key, Value>> iterator = scanner.iterator(); + int count = 0; + while (iterator.hasNext()) { + iterator.next(); + count++; + } + log.info(""); + log.info("Total rows copied: " + count); + log.info(""); + + log.info("Demo done"); + } +} http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/71b5b1c9/extras/rya.merger/src/test/java/mvm/rya/accumulo/mr/merge/demo/MergeToolDemo.java ---------------------------------------------------------------------- diff --git a/extras/rya.merger/src/test/java/mvm/rya/accumulo/mr/merge/demo/MergeToolDemo.java b/extras/rya.merger/src/test/java/mvm/rya/accumulo/mr/merge/demo/MergeToolDemo.java new file mode 100644 index 0000000..6a8e779 --- /dev/null +++ b/extras/rya.merger/src/test/java/mvm/rya/accumulo/mr/merge/demo/MergeToolDemo.java @@ -0,0 +1,303 @@ +package mvm.rya.accumulo.mr.merge.demo; + +import static mvm.rya.accumulo.mr.merge.util.TestUtils.createRyaStatement; +import static mvm.rya.accumulo.mr.merge.util.ToolConfigUtils.makeArgument; + +import java.util.Date; +import java.util.Iterator; +import java.util.Map.Entry; +import java.util.concurrent.TimeUnit; + +import org.apache.accumulo.core.client.Scanner; +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.Value; +import org.apache.accumulo.core.security.ColumnVisibility; +import org.apache.log4j.Logger; + +import mvm.rya.accumulo.AccumuloRdfConfiguration; +import mvm.rya.accumulo.AccumuloRyaDAO; +import mvm.rya.accumulo.mr.merge.CopyTool; +import mvm.rya.accumulo.mr.merge.MergeTool; +import mvm.rya.accumulo.mr.merge.demo.util.DemoUtilities; +import mvm.rya.accumulo.mr.merge.demo.util.DemoUtilities.LoggingDetail; +import mvm.rya.accumulo.mr.merge.driver.AccumuloDualInstanceDriver; +import mvm.rya.accumulo.mr.merge.util.AccumuloRyaUtils; +import mvm.rya.accumulo.mr.merge.util.TestUtils; +import mvm.rya.accumulo.mr.merge.util.TimeUtils; +import mvm.rya.accumulo.mr.utils.MRUtils; +import mvm.rya.api.RdfCloudTripleStoreConstants; +import mvm.rya.api.domain.RyaStatement; + +/** + * Tests for {@link MergeTool}. + */ +public class MergeToolDemo { + private static final Logger log = Logger.getLogger(MergeToolDemo.class); + + private static final boolean IS_MOCK = true; + private static final boolean USE_TIME_SYNC = false; + private static final boolean USE_MERGE_FILE_INPUT = false; + private static final boolean IS_PROMPTING_ENABLED = true; + private static final boolean IS_START_TIME_DIALOG_ENABLED = true; + private static final LoggingDetail LOGGING_DETAIL = LoggingDetail.LIGHT; + + private static final String CHILD_SUFFIX = MergeTool.CHILD_SUFFIX; + + private static final String PARENT_PASSWORD = AccumuloDualInstanceDriver.PARENT_PASSWORD; + private static final String PARENT_INSTANCE = AccumuloDualInstanceDriver.PARENT_INSTANCE; + private static final String PARENT_TABLE_PREFIX = AccumuloDualInstanceDriver.PARENT_TABLE_PREFIX; + private static final String PARENT_AUTH = AccumuloDualInstanceDriver.PARENT_AUTH; + private static final ColumnVisibility PARENT_COLUMN_VISIBILITY = new ColumnVisibility(PARENT_AUTH); + private static final String PARENT_TOMCAT_URL = "http://rya-example-box:8080";; + + private static final String CHILD_PASSWORD = AccumuloDualInstanceDriver.CHILD_PASSWORD; + private static final String CHILD_INSTANCE = AccumuloDualInstanceDriver.CHILD_INSTANCE; + private static final String CHILD_TABLE_PREFIX = AccumuloDualInstanceDriver.CHILD_TABLE_PREFIX; + private static final String CHILD_AUTH = AccumuloDualInstanceDriver.CHILD_AUTH; + private static final ColumnVisibility CHILD_COLUMN_VISIBILITY = new ColumnVisibility(CHILD_AUTH); + private static final String CHILD_TOMCAT_URL = "http://localhost:8080";; + + + private static final long MERGE_FILE_TIME = 1463691894954L; + private static final Date TODAY = USE_MERGE_FILE_INPUT ? new Date(MERGE_FILE_TIME) : new Date(System.currentTimeMillis()); + private static final Date YESTERDAY = TestUtils.dayBefore(TODAY); + private static final Date LAST_MONTH = TestUtils.monthBefore(TODAY); + + private AccumuloRyaDAO parentDao; + private AccumuloRyaDAO childDao; + + private AccumuloRdfConfiguration parentConfig; + private AccumuloRdfConfiguration childConfig; + + private AccumuloDualInstanceDriver accumuloDualInstanceDriver; + + public static void main(String args[]) { + DemoUtilities.setupLogging(LOGGING_DETAIL); + log.info("Setting up Merge Tool Demo"); + + Thread.setDefaultUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { + @Override + public void uncaughtException(Thread thread, Throwable throwable) { + log.fatal("Uncaught exception in " + thread.getName(), throwable); + } + }); + + final MergeToolDemo mergeToolDemo = new MergeToolDemo(); + Runtime.getRuntime().addShutdownHook(new Thread() { + @Override + public void run() { + log.info("Shutting down..."); + try { + mergeToolDemo.tearDown(); + } catch (Exception e) { + log.error("Error while shutting down", e); + } finally { + log.info("Done shutting down"); + } + } + }); + + try { + mergeToolDemo.setUp(); + mergeToolDemo.testMergeTool(); + } catch (Exception e) { + log.error("Error while testing merge tool", e); + } finally { + try { + mergeToolDemo.tearDown(); + } catch (Exception e) { + log.error("Error shutting down merge tool.", e); + } + } + + System.exit(0); + } + + public void setUp() throws Exception { + accumuloDualInstanceDriver = new AccumuloDualInstanceDriver(IS_MOCK, false, false, true, !USE_MERGE_FILE_INPUT); + accumuloDualInstanceDriver.setUpInstances(); + + accumuloDualInstanceDriver.setUpTables(); + + accumuloDualInstanceDriver.setUpDaos(); + + accumuloDualInstanceDriver.setUpConfigs(); + + parentConfig = accumuloDualInstanceDriver.getParentConfig(); + childConfig = accumuloDualInstanceDriver.getChildConfig(); + parentDao = accumuloDualInstanceDriver.getParentDao(); + childDao = accumuloDualInstanceDriver.getChildDao(); + } + + public void tearDown() throws Exception { + log.info("Tearing down..."); + accumuloDualInstanceDriver.tearDown(); + } + + private void mergeToolRun(Date startDate) { + MergeTool.setupAndRun(new String[] { + makeArgument(MRUtils.AC_MOCK_PROP, Boolean.toString(IS_MOCK)), + makeArgument(MRUtils.AC_INSTANCE_PROP, PARENT_INSTANCE), + makeArgument(MRUtils.AC_USERNAME_PROP, accumuloDualInstanceDriver.getParentUser()), + makeArgument(MRUtils.AC_PWD_PROP, PARENT_PASSWORD), + makeArgument(MRUtils.TABLE_PREFIX_PROPERTY, PARENT_TABLE_PREFIX), + makeArgument(MRUtils.AC_AUTH_PROP, PARENT_AUTH), + makeArgument(MRUtils.AC_ZK_PROP, accumuloDualInstanceDriver.getParentZooKeepers()), + makeArgument(CopyTool.PARENT_TOMCAT_URL_PROP, PARENT_TOMCAT_URL), + makeArgument(MRUtils.AC_MOCK_PROP + CHILD_SUFFIX, Boolean.toString(IS_MOCK)), + makeArgument(MRUtils.AC_INSTANCE_PROP + CHILD_SUFFIX, CHILD_INSTANCE), + makeArgument(MRUtils.AC_USERNAME_PROP + CHILD_SUFFIX, accumuloDualInstanceDriver.getChildUser()), + makeArgument(MRUtils.AC_PWD_PROP + CHILD_SUFFIX, CHILD_PASSWORD), + makeArgument(MRUtils.TABLE_PREFIX_PROPERTY + CHILD_SUFFIX, CHILD_TABLE_PREFIX), + makeArgument(MRUtils.AC_AUTH_PROP + CHILD_SUFFIX, CHILD_AUTH), + makeArgument(MRUtils.AC_ZK_PROP + CHILD_SUFFIX, accumuloDualInstanceDriver.getChildZooKeepers()), + makeArgument(CopyTool.CHILD_TOMCAT_URL_PROP, CHILD_TOMCAT_URL), + makeArgument(CopyTool.NTP_SERVER_HOST_PROP, TimeUtils.DEFAULT_TIME_SERVER_HOST), + makeArgument(CopyTool.USE_NTP_SERVER_PROP, Boolean.toString(USE_TIME_SYNC)), + makeArgument(MergeTool.USE_MERGE_FILE_INPUT, Boolean.toString(USE_MERGE_FILE_INPUT)), + makeArgument(MergeTool.MERGE_FILE_INPUT_PATH, "resources/test/merge_tool_file_input/"), + makeArgument(MergeTool.START_TIME_PROP, MergeTool.getStartTimeString(startDate, IS_START_TIME_DIALOG_ENABLED)) + }); + + log.info("Finished running tool."); + } + + public void testMergeTool() throws Exception { + log.info(""); + StringBuilder sb = new StringBuilder(); + sb.append("Cases to check\n"); + sb.append("\n"); + sb.append("| **case** | **Parent** | **Child** | **assume that** | **merge modification** | **in parent after** |\n"); + sb.append("|----------|------------|-----------|-----------------|------------------------|---------------------|\n"); + sb.append("| 1 | older | missing | child deleted | delete from parent | no |\n"); + sb.append("| 2 | newer | missing | parent added | do nothing | yes |\n"); + sb.append("| 3 | missing | older | parent deleted | do nothing | no |\n"); + sb.append("| 4 | missing | newer | child added | add to parent | yes |\n"); + sb.append("| 5x | older | older | same key | do nothing | yes |\n"); + sb.append("| 6x | newer | newer | same key | do nothing | yes |\n"); + sb.append("| 7* | older | newer | child updated | do cases 1 and 4 | yes |\n"); + sb.append("| 8* | newer | older | parent updated | do cases 2 and 3 | yes |\n"); + sb.append("\n"); + sb.append("x - The two cases having the same key and timestamp are already merged.\n"); + sb.append("* - The last two cases are really two distinct keys, since the timestamp is part of the Accumulo key. They are handled as two comparisons. They are different versions of the same key."); + sb.append("\n"); + log.info(sb.toString()); + + log.info("Additional Case 9 to combine column visibilities of same rows.\n"); + + log.info("Initial state of parent and child SPO tables..."); + + DemoUtilities.promptEnterKey(IS_PROMPTING_ENABLED); + + // This statement was in both parent/child instances a month ago but after the start time of yesterday + // the child deleted it and the parent still has it. It should be deleted from the parent after merging. + RyaStatement ryaStatementCase1 = createRyaStatement("c1", "parent older", "child missing", LAST_MONTH); + + // This statement was added by the parent after the start time of yesterday and doesn't exist in the child. + // It should stay in the parent after merging. + RyaStatement ryaStatementCase2 = createRyaStatement("c2", "parent newer", "child missing", TODAY); + + // This statement was in both parent/child instances a month ago but after the start time of yesterday + // the parent deleted it and the child still has it. It should stay deleted in the parent after merging. + RyaStatement ryaStatementCase3 = createRyaStatement("c3", "parent missing", "child older", LAST_MONTH); + + // This statement was added by the child after the start time of yesterday and doesn't exist in the parent. + // It should be added to the parent after merging. + RyaStatement ryaStatementCase4 = createRyaStatement("c4", "parent missing", "child newer", TODAY); + + // This statement was in both parent/child instances a month ago and is before the start time of yesterday + // but it was left alone. It should remain in the parent after merging. + RyaStatement ryaStatementCase5 = createRyaStatement("c5", "parent older", "child older", LAST_MONTH); + + // This statement was added by the parent and child after the start time of yesterday. + // It should remain in the parent after merging. + RyaStatement ryaStatementCase6 = createRyaStatement("c6", "parent newer", "child newer", TODAY); + + // This statement was modified by the child after the start time of yesterday. + // It should deleted and then re-added to the parent with the new child data after merging. + RyaStatement ryaStatementCase7 = createRyaStatement("c7", "parent older", "child newer", LAST_MONTH); + + // This statement was modified by the parent after the start time of yesterday. + // It should deleted and then re-added to the parent with the new child data after merging. + RyaStatement ryaStatementCase8 = createRyaStatement("c8", "parent newer", "child older", LAST_MONTH); + + // This statement was modified by the child to change the column visibility. + // The parent should combine the child's visibility with its visibility. + RyaStatement ryaStatementVisibilityDifferent = createRyaStatement("c9", "I see", "you", LAST_MONTH); + ryaStatementVisibilityDifferent.setColumnVisibility(PARENT_COLUMN_VISIBILITY.getExpression()); + + // Change statements that were updated by the parent or child after the start time. + RyaStatement ryaStatementCase7Updated = TestUtils.copyRyaStatement(ryaStatementCase7); + ryaStatementCase7Updated.setSubject(TestUtils.createRyaUri("c7 BOB")); + ryaStatementCase7Updated.setTimestamp(TODAY.getTime()); + + RyaStatement ryaStatementCase8Updated = TestUtils.copyRyaStatement(ryaStatementCase8); + ryaStatementCase8Updated.setSubject(TestUtils.createRyaUri("c8 SUSAN")); + ryaStatementCase8Updated.setTimestamp(TODAY.getTime()); + + RyaStatement ryaStatementVisibilityDifferentUpdated = TestUtils.copyRyaStatement(ryaStatementVisibilityDifferent); + ryaStatementVisibilityDifferentUpdated.setColumnVisibility(CHILD_COLUMN_VISIBILITY.getExpression()); + + // Setup initial parent instance with 7 rows + // This is the state of the parent data (as it is today) before merging occurs which will use the specified start time of yesterday. + parentDao.add(ryaStatementVisibilityDifferent); // Merging should update statement + parentDao.add(ryaStatementCase1); // Merging should delete statement + parentDao.add(ryaStatementCase2); // Merging should keep statement + parentDao.add(ryaStatementCase5); // Merging should keep statement + parentDao.add(ryaStatementCase6); // Merging should keep statement + parentDao.add(ryaStatementCase7); // Merging should update statement + parentDao.add(ryaStatementCase8Updated); // Merging should keep statement + + ryaStatementVisibilityDifferent.setColumnVisibility(CHILD_COLUMN_VISIBILITY.getExpression()); + + if (!USE_MERGE_FILE_INPUT) { + // Simulate the child coming back with a modified data set before the merging occurs. + // There should be 7 rows in the child instance. + childDao.add(ryaStatementVisibilityDifferentUpdated); + childDao.add(ryaStatementCase3); + childDao.add(ryaStatementCase4); + childDao.add(ryaStatementCase5); + childDao.add(ryaStatementCase6); // Merging should add statement + childDao.add(ryaStatementCase7Updated); + childDao.add(ryaStatementCase8); // Merging should add statement + + // Create 5 hour 5 minute 5 second and 5 millisecond offset metadata key + Long timeOffset = TimeUnit.HOURS.toMillis(5) + TimeUnit.MINUTES.toMillis(5) + TimeUnit.SECONDS.toMillis(5) + TimeUnit.MILLISECONDS.toMillis(5); + log.info("Creating parent time offset of: " + TimeUtils.getDurationBreakdown(timeOffset)); + AccumuloRyaUtils.setTimeOffset(timeOffset, childDao); + } + + log.info("7 rows in parent before merging."); + + AccumuloRyaUtils.printTablePretty(PARENT_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, parentConfig); + if (!USE_MERGE_FILE_INPUT) { + AccumuloRyaUtils.printTablePretty(CHILD_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, childConfig); + } + + DemoUtilities.promptEnterKey(IS_PROMPTING_ENABLED); + + log.info("Starting merge tool. Merging all data after the specified start time: " + YESTERDAY); + + mergeToolRun(YESTERDAY); + + +// for (String tableSuffix : AccumuloDualInstanceDriver.TABLE_NAME_SUFFIXES) { +// AccumuloRyaUtils.printTablePretty(PARENT_TABLE_PREFIX + tableSuffix, parentConfig); +// } + + AccumuloRyaUtils.printTablePretty(PARENT_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, parentConfig); + + + Scanner scanner = AccumuloRyaUtils.getScanner(PARENT_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, parentConfig); + Iterator<Entry<Key, Value>> iterator = scanner.iterator(); + int count = 0; + while (iterator.hasNext()) { + iterator.next(); + count++; + } + // Make sure we have all of them in the parent. + log.info("Total rows in parent: " + count); + + log.info("Demo done"); + } +}
