TableUtils { public static synchronized void createTable(final AccumuloStore store) throws StoreException, TableExistsException { final String tableName = store.getTableName(); if (null == tableName) { throw new AccumuloRuntimeException("Table name is required."); } final Connector connector = store.getConnection(); if (connector.tableOperations().exists(tableName)) { LOGGER.info("Table {} exists, not creating", tableName); return; } try { final String namespace = store.getProperties().getNamespace(); if (StringUtils.isNotBlank(namespace)) { if (!connector.namespaceOperations().exists(namespace)) { LOGGER.info("Creating namespace {} as user {}", namespace, connector.whoami()); try { connector.namespaceOperations().create(namespace); } catch (final NamespaceExistsException e) { } } } LOGGER.info("Creating table {} as user {}", tableName, connector.whoami()); connector.tableOperations().create(tableName); final String repFactor = store.getProperties().getTableFileReplicationFactor(); if (null != repFactor) { LOGGER.info("Table file replication set to {} on table {}", repFactor, tableName); connector.tableOperations().setProperty(tableName, Property.TABLE_FILE_REPLICATION.getKey(), repFactor); } LOGGER.info("Enabling Bloom filter on table {}", tableName); connector.tableOperations().setProperty(tableName, Property.TABLE_BLOOM_ENABLED.getKey(), "true"); connector.tableOperations().setProperty(tableName, Property.TABLE_BLOOM_KEY_FUNCTOR.getKey(), store.getKeyPackage().getKeyFunctor().getClass().getName()); LOGGER.info("Removing versioning iterator from table {}", tableName); final EnumSet<IteratorScope> iteratorScopes = EnumSet.allOf(IteratorScope.class); connector.tableOperations().removeIterator(tableName, "vers", iteratorScopes); if (store.getSchema().isAggregationEnabled()) { LOGGER.info("Adding Aggregator iterator to table {} for all scopes", tableName); connector.tableOperations().attachIterator(tableName, store.getKeyPackage().getIteratorFactory().getAggregatorIteratorSetting(store)); } else { LOGGER.info("Aggregator iterator has not been added to table {}", tableName); } if (store.getProperties().getEnableValidatorIterator()) { final IteratorSetting itrSetting = store.getKeyPackage().getIteratorFactory().getValidatorIteratorSetting(store); if (null == itrSetting) { LOGGER.info("Not adding Validator iterator to table {} as there are no validation functions defined in the schema", tableName); } else { LOGGER.info("Adding Validator iterator to table {} for all scopes", tableName); connector.tableOperations().attachIterator(tableName, store.getKeyPackage().getIteratorFactory().getValidatorIteratorSetting(store)); } } else { LOGGER.info("Validator iterator has not been added to table {}", tableName); } } catch (final AccumuloSecurityException | TableNotFoundException | AccumuloException | IteratorSettingException e) { throw new StoreException(e.getMessage(), e); } setLocalityGroups(store); } private  TableUtils(); static void ensureTableExists(final AccumuloStore store); static synchronized void createTable(final AccumuloStore store); static void setLocalityGroups(final AccumuloStore store); static BatchWriter createBatchWriter(final AccumuloStore store); static Connector getConnector(final String instanceName, final String zookeepers, final String userName,
                                         final String password); static Authorizations getCurrentAuthorizations(final Connector connection); static final String COLUMN_FAMILIES_OPTION; }
@Test public void shouldCreateTableWithCorrectLocalityGroups() throws Exception { final AccumuloStore store = new SingleUseAccumuloStore(); final Schema schema = new Schema.Builder() .type(TestTypes.ID_STRING, String.class) .type(TestTypes.DIRECTED_TRUE, Boolean.class) .edge(TestGroups.EDGE, new SchemaEdgeDefinition.Builder() .source(TestTypes.ID_STRING) .destination(TestTypes.ID_STRING) .directed(TestTypes.DIRECTED_TRUE) .build()) .build(); store.initialise(LOCALITY_GRAPH_ID, schema, PROPERTIES); TableUtils.createTable(store); final Map<String, Set<Text>> localityGroups = store.getConnection().tableOperations().getLocalityGroups(LOCALITY_GRAPH_ID); assertEquals(1, localityGroups.size()); Set<Text> localityGroup = localityGroups.get(TestGroups.EDGE); assertEquals(1, localityGroup.size()); assertEquals(new Text(TestGroups.EDGE), localityGroup.toArray()[0]); }
@Test public void shouldCreateTableCorrectlyIfSchemaContainsNoAggregators() throws Exception { final AccumuloStore store = new SingleUseAccumuloStore(); final Schema schema = new Schema.Builder() .type(TestTypes.ID_STRING, new TypeDefinition.Builder() .clazz(String.class) .validateFunctions(new Exists()) .build()) .type(TestTypes.DIRECTED_TRUE, Boolean.class) .edge(TestGroups.EDGE, new SchemaEdgeDefinition.Builder() .source(TestTypes.ID_STRING) .destination(TestTypes.ID_STRING) .directed(TestTypes.DIRECTED_TRUE) .aggregate(false) .build()) .build(); store.initialise(NO_AGGREGATORS_GRAPH_ID, schema, PROPERTIES); TableUtils.createTable(store); final Map<String, EnumSet<IteratorScope>> itrs = store.getConnection().tableOperations().listIterators(NO_AGGREGATORS_GRAPH_ID); assertEquals(1, itrs.size()); final EnumSet<IteratorScope> validator = itrs.get(AccumuloStoreConstants.VALIDATOR_ITERATOR_NAME); assertEquals(EnumSet.allOf(IteratorScope.class), validator); final IteratorSetting validatorSetting = store.getConnection().tableOperations().getIteratorSetting(NO_AGGREGATORS_GRAPH_ID, AccumuloStoreConstants.VALIDATOR_ITERATOR_NAME, IteratorScope.majc); assertEquals(AccumuloStoreConstants.VALIDATOR_ITERATOR_PRIORITY, validatorSetting.getPriority()); assertEquals(ValidatorFilter.class.getName(), validatorSetting.getIteratorClass()); final Map<String, String> validatorOptions = validatorSetting.getOptions(); assertNotNull(Schema.fromJson(validatorOptions.get(AccumuloStoreConstants.SCHEMA).getBytes(CommonConstants.UTF_8)).getEdge(TestGroups.EDGE)); assertEquals(ByteEntityAccumuloElementConverter.class.getName(), validatorOptions.get(AccumuloStoreConstants.ACCUMULO_ELEMENT_CONVERTER_CLASS)); final EnumSet<IteratorScope> aggregator = itrs.get(AccumuloStoreConstants.AGGREGATOR_ITERATOR_NAME); assertNull(aggregator); final IteratorSetting aggregatorSetting = store.getConnection().tableOperations().getIteratorSetting(NO_AGGREGATORS_GRAPH_ID, AccumuloStoreConstants.AGGREGATOR_ITERATOR_NAME, IteratorScope.majc); assertNull(aggregatorSetting); final Map<String, String> tableProps = new HashMap<>(); for (final Map.Entry<String, String> entry : store.getConnection() .tableOperations().getProperties(NO_AGGREGATORS_GRAPH_ID)) { tableProps.put(entry.getKey(), entry.getValue()); } assertEquals(0, Integer.parseInt(tableProps.get(Property.TABLE_FILE_REPLICATION.getKey()))); }
@Test public void shouldThrowExceptionIfTableNameIsNotSpecifiedWhenCreatingTable() throws StoreException, TableExistsException { final Schema schema = new Schema.Builder() .type("int", Integer.class) .type("string", String.class) .type("boolean", Boolean.class) .edge("EDGE", new SchemaEdgeDefinition.Builder() .source("string") .destination("string") .directed("boolean") .build()) .build(); final AccumuloProperties properties = new AccumuloProperties(); properties.setStoreClass(PROPERTIES.getStoreClass()); properties.setZookeepers((PROPERTIES.getZookeepers())); final AccumuloStore store = new AccumuloStore(); assertThrows(IllegalArgumentException.class, () -> store.initialise(null, schema, properties)); assertThrows(AccumuloRuntimeException.class, () -> TableUtils.createTable(store)); }