CopyTool extends AbstractDualInstanceAccumuloMRTool { public AccumuloInstanceDriver getChildAccumuloInstanceDriver() { return childAccumuloInstanceDriver; }  void setup(); @Override int run(final String[] strings); void createTableIfNeeded(final String childTableName); static Path getPath(final String first, final String... more); void importFilesToChildTable(final String childTableName); void copyLocalToHdfs(final Path localInputPath, final Path hdfsOutputPath); static void copyLocalToHdfs(final Path localInputPath, final Path hdfsOutputPath, final Configuration configuration); void copyHdfsToLocal(final Path hdfsInputPath, final Path localOutputPath); static void copyHdfsToLocal(final Path hdfsInputPath, final Path localOutputPath, final Configuration configuration); int setupAndRun(final String[] args); static void main(final String[] args); static IteratorSetting getStartTimeSetting(final String startTimeString); static IteratorSetting getStartTimeSetting(final Date date); static IteratorSetting getStartTimeSetting(final long time); AccumuloInstanceDriver getChildAccumuloInstanceDriver(); void shutdown(); static final String COPY_TABLE_LIST_PROP; static final String CREATE_CHILD_INSTANCE_TYPE_PROP; static final String PARENT_TIME_OFFSET_PROP; static final String CHILD_TIME_OFFSET_PROP; static final String NTP_SERVER_HOST_PROP; static final String PARENT_TOMCAT_URL_PROP; static final String CHILD_TOMCAT_URL_PROP; static final String COPY_RUN_TIME_PROP; static final String USE_NTP_SERVER_PROP; static final String USE_COPY_FILE_OUTPUT; static final String COPY_FILE_OUTPUT_PATH; static final String COPY_FILE_OUTPUT_COMPRESSION_TYPE; static final String USE_COPY_FILE_OUTPUT_DIRECTORY_CLEAR; static final String COPY_FILE_IMPORT_DIRECTORY; static final String USE_COPY_FILE_IMPORT; static final String USE_COPY_QUERY_SPARQL; static final String QUERY_STRING_PROP; static final String QUERY_FILE_PROP; }
@Test public void testCopyTool() throws Exception { final RyaStatement ryaStatementOutOfTimeRange = createRyaStatement("coach", "called", "timeout", LAST_MONTH); final RyaStatement ryaStatementShouldCopy1 = createRyaStatement("bob", "catches", "ball", YESTERDAY); final RyaStatement ryaStatementShouldCopy2 = createRyaStatement("bill", "talks to", "john", YESTERDAY); final RyaStatement ryaStatementShouldCopy3 = createRyaStatement("susan", "eats", "burgers", TODAY); final RyaStatement ryaStatementShouldCopy4 = createRyaStatement("ronnie", "plays", "guitar", TODAY); final RyaStatement ryaStatementDoesNotExist1 = createRyaStatement("nobody", "was", "here", LAST_MONTH); final RyaStatement ryaStatementDoesNotExist2 = createRyaStatement("statement", "not", "found", YESTERDAY); final RyaStatement ryaStatementDoesNotExist3 = createRyaStatement("key", "does not", "exist", TODAY); final RyaStatement ryaStatementVisibilityDifferent = createRyaStatement("I", "see", "you", YESTERDAY); ryaStatementVisibilityDifferent.setColumnVisibility(PARENT_COLUMN_VISIBILITY.getExpression()); parentDao.add(ryaStatementOutOfTimeRange); parentDao.add(ryaStatementShouldCopy1); parentDao.add(ryaStatementShouldCopy2); parentDao.add(ryaStatementShouldCopy3); parentDao.add(ryaStatementShouldCopy4); parentDao.add(ryaStatementVisibilityDifferent); AccumuloRyaUtils.printTable(PARENT_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, parentConfig); log.info("Starting copy tool. Copying all data after the specified start time: " + YESTERDAY); isImporting = false; copyToolRun(YESTERDAY); final String childUser = accumuloDualInstanceDriver.getChildUser(); final Connector childConnector = ConfigUtils.getConnector(childConfig); accumuloDualInstanceDriver.getChildAccumuloInstanceDriver().setConnector(childConnector); accumuloDualInstanceDriver.getChildAccumuloInstanceDriver().setUpTables(); accumuloDualInstanceDriver.getChildAccumuloInstanceDriver().setUpDao(); childDao = accumuloDualInstanceDriver.getChildDao(); final SecurityOperations childSecOps = accumuloDualInstanceDriver.getChildSecOps(); Authorizations newChildAuths = AccumuloRyaUtils.addUserAuths(childUser, childSecOps, PARENT_AUTH); childSecOps.changeUserAuthorizations(childUser, newChildAuths); final String childAuthString = newChildAuths.toString(); final List<String> duplicateKeys = MergeTool.DUPLICATE_KEY_MAP.get(MRUtils.AC_AUTH_PROP); childConfig.set(MRUtils.AC_AUTH_PROP, childAuthString); for (final 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); final Scanner scanner = AccumuloRyaUtils.getScanner(CHILD_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, childConfig); final Iterator<Entry<Key, Value>> iterator = scanner.iterator(); int count = 0; while (iterator.hasNext()) { iterator.next(); count++; } assertEquals(5, count); assertStatementInChild("Child included statement that was out of time range", 0, ryaStatementOutOfTimeRange); assertStatementInChild("Child missing statement 1 that was in parent", 1, ryaStatementShouldCopy1); assertStatementInChild("Child missing statement 2 that was in parent", 1, ryaStatementShouldCopy2); assertStatementInChild("Child missing statement 3 that was in parent", 1, ryaStatementShouldCopy3); assertStatementInChild("Child missing statement 4 that was in parent", 1, ryaStatementShouldCopy4); assertStatementInChild("Child included statement 1 that was not in parent", 0, ryaStatementDoesNotExist1); assertStatementInChild("Child included statement 2 that was not in parent", 0, ryaStatementDoesNotExist2); assertStatementInChild("Child included statement 3 that was not in parent", 0, ryaStatementDoesNotExist3); assertStatementInChild("Child missing statement with child visibility", 1, ryaStatementVisibilityDifferent); childConfig.set(RdfCloudTripleStoreConfiguration.CONF_QUERY_AUTH, PARENT_AUTH); final SecurityOperations secOps = IS_MOCK ? accumuloDualInstanceDriver.getChildSecOps() : childSecOps; newChildAuths = AccumuloRyaUtils.addUserAuths(accumuloDualInstanceDriver.getChildUser(), secOps, PARENT_AUTH); secOps.changeUserAuthorizations(accumuloDualInstanceDriver.getChildUser(), newChildAuths); assertStatementInChild("Child missing statement with parent visibility", 1, ryaStatementVisibilityDifferent); childConfig.set(RdfCloudTripleStoreConfiguration.CONF_QUERY_AUTH, "bad_auth"); final CloseableIteration<RyaStatement, RyaDAOException> iter = childDao.getQueryEngine().query(ryaStatementVisibilityDifferent, childConfig); count = 0; try { while (iter.hasNext()) { iter.next(); count++; } } catch (final Exception e) { if (!(e.getCause() instanceof AccumuloSecurityException)) { fail(); } } iter.close(); assertEquals(0, count); childConfig.set(RdfCloudTripleStoreConfiguration.CONF_QUERY_AUTH, childAuthString); log.info("DONE"); }
@Test public void testImportDirectoryTool() throws Exception { log.info(""); log.info("Setting up initial state of parent before importing directory to child..."); log.info("Adding data to parent..."); log.info("Starting import directory tool. Importing all data after the specified start time: " + YESTERDAY); log.info(""); isImporting = true; copyToolRun(YESTERDAY); final String childUser = accumuloDualInstanceDriver.getChildUser(); final Connector childConnector = ConfigUtils.getConnector(childConfig); accumuloDualInstanceDriver.getChildAccumuloInstanceDriver().setConnector(childConnector); accumuloDualInstanceDriver.getChildAccumuloInstanceDriver().setUpTables(); accumuloDualInstanceDriver.getChildAccumuloInstanceDriver().setUpDao(); final SecurityOperations childSecOps = accumuloDualInstanceDriver.getChildSecOps(); final Authorizations newChildAuths = AccumuloRyaUtils.addUserAuths(childUser, childSecOps, PARENT_AUTH); childSecOps.changeUserAuthorizations(childUser, newChildAuths); final String childAuthString = newChildAuths.toString(); final List<String> duplicateKeys = MergeTool.DUPLICATE_KEY_MAP.get(MRUtils.AC_AUTH_PROP); childConfig.set(MRUtils.AC_AUTH_PROP, childAuthString); for (final String key : duplicateKeys) { childConfig.set(key, childAuthString); } AccumuloRyaUtils.printTablePretty(CHILD_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, childConfig); final Scanner scanner = AccumuloRyaUtils.getScanner(CHILD_TABLE_PREFIX + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, childConfig); final Iterator<Entry<Key, Value>> iterator = scanner.iterator(); int count = 0; while (iterator.hasNext()) { iterator.next(); count++; } log.info(""); log.info("Total rows imported: " + count); log.info(""); assertEquals(20, count); log.info("DONE"); }