http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/44a2dcf0/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/AccumuloRyaQueryEngine.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/AccumuloRyaQueryEngine.java b/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/AccumuloRyaQueryEngine.java new file mode 100644 index 0000000..ba3ffd2 --- /dev/null +++ b/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/AccumuloRyaQueryEngine.java @@ -0,0 +1,410 @@ +package mvm.rya.accumulo.query; + +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + */ + + + +import static mvm.rya.api.RdfCloudTripleStoreUtils.layoutToTable; +import info.aduna.iteration.CloseableIteration; + +import java.io.IOException; +import java.util.Collection; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.Map; + +import mvm.rya.accumulo.AccumuloRdfConfiguration; +import mvm.rya.api.RdfCloudTripleStoreConfiguration; +import mvm.rya.api.RdfCloudTripleStoreConstants; +import mvm.rya.api.RdfCloudTripleStoreConstants.TABLE_LAYOUT; +import mvm.rya.api.RdfCloudTripleStoreUtils; +import mvm.rya.api.domain.RyaRange; +import mvm.rya.api.domain.RyaStatement; +import mvm.rya.api.domain.RyaType; +import mvm.rya.api.domain.RyaURI; +import mvm.rya.api.layout.TableLayoutStrategy; +import mvm.rya.api.persist.RyaDAOException; +import mvm.rya.api.persist.query.BatchRyaQuery; +import mvm.rya.api.persist.query.RyaQuery; +import mvm.rya.api.persist.query.RyaQueryEngine; +import mvm.rya.api.query.strategy.ByteRange; +import mvm.rya.api.query.strategy.TriplePatternStrategy; +import mvm.rya.api.resolver.RyaContext; +import mvm.rya.api.resolver.RyaTripleContext; +import mvm.rya.api.resolver.triple.TripleRowRegex; +import mvm.rya.api.utils.CloseableIterableIteration; + +import org.apache.accumulo.core.client.BatchScanner; +import org.apache.accumulo.core.client.Connector; +import org.apache.accumulo.core.client.IteratorSetting; +import org.apache.accumulo.core.client.Scanner; +import org.apache.accumulo.core.client.ScannerBase; +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.Range; +import org.apache.accumulo.core.data.Value; +import org.apache.accumulo.core.iterators.user.RegExFilter; +import org.apache.accumulo.core.iterators.user.TimestampFilter; +import org.apache.accumulo.core.security.Authorizations; +import org.apache.hadoop.io.Text; +import org.calrissian.mango.collect.CloseableIterable; +import org.calrissian.mango.collect.CloseableIterables; +import org.calrissian.mango.collect.FluentCloseableIterable; +import org.openrdf.query.BindingSet; + +import com.google.common.base.Function; +import com.google.common.base.Preconditions; +import com.google.common.collect.FluentIterable; +import com.google.common.collect.Iterators; + +/** + * Date: 7/17/12 + * Time: 9:28 AM + */ +public class AccumuloRyaQueryEngine implements RyaQueryEngine<AccumuloRdfConfiguration> { + + private AccumuloRdfConfiguration configuration; + private Connector connector; + private RyaTripleContext ryaContext; + private final Map<TABLE_LAYOUT, KeyValueToRyaStatementFunction> keyValueToRyaStatementFunctionMap = new HashMap<TABLE_LAYOUT, KeyValueToRyaStatementFunction>(); + + public AccumuloRyaQueryEngine(Connector connector) { + this(connector, new AccumuloRdfConfiguration()); + } + + public AccumuloRyaQueryEngine(Connector connector, AccumuloRdfConfiguration conf) { + this.connector = connector; + this.configuration = conf; + ryaContext = RyaTripleContext.getInstance(conf); + keyValueToRyaStatementFunctionMap.put(TABLE_LAYOUT.SPO, new KeyValueToRyaStatementFunction(TABLE_LAYOUT.SPO, ryaContext)); + keyValueToRyaStatementFunctionMap.put(TABLE_LAYOUT.PO, new KeyValueToRyaStatementFunction(TABLE_LAYOUT.PO, ryaContext)); + keyValueToRyaStatementFunctionMap.put(TABLE_LAYOUT.OSP, new KeyValueToRyaStatementFunction(TABLE_LAYOUT.OSP, ryaContext)); + } + + @Override + public CloseableIteration<RyaStatement, RyaDAOException> query(RyaStatement stmt, AccumuloRdfConfiguration conf) throws RyaDAOException { + if (conf == null) { + conf = configuration; + } + + RyaQuery ryaQuery = RyaQuery.builder(stmt).load(conf).build(); + CloseableIterable<RyaStatement> results = query(ryaQuery); + + return new CloseableIterableIteration<RyaStatement, RyaDAOException>(results); + } + + protected String getData(RyaType ryaType) { + return (ryaType != null) ? (ryaType.getData()) : (null); + } + + @Override + public CloseableIteration<? extends Map.Entry<RyaStatement, BindingSet>, RyaDAOException> queryWithBindingSet(Collection<Map.Entry<RyaStatement, BindingSet>> stmts, AccumuloRdfConfiguration conf) throws RyaDAOException { + if (conf == null) { + conf = configuration; + } + //query configuration + Authorizations authorizations = conf.getAuthorizations(); + Long ttl = conf.getTtl(); + Long maxResults = conf.getLimit(); + Integer maxRanges = conf.getMaxRangesForScanner(); + Integer numThreads = conf.getNumThreads(); + + //TODO: cannot span multiple tables here + try { + Collection<Range> ranges = new HashSet<Range>(); + RangeBindingSetEntries rangeMap = new RangeBindingSetEntries(); + TABLE_LAYOUT layout = null; + RyaURI context = null; + TriplePatternStrategy strategy = null; + for (Map.Entry<RyaStatement, BindingSet> stmtbs : stmts) { + RyaStatement stmt = stmtbs.getKey(); + context = stmt.getContext(); //TODO: This will be overwritten + BindingSet bs = stmtbs.getValue(); + strategy = ryaContext.retrieveStrategy(stmt); + if (strategy == null) { + throw new IllegalArgumentException("TriplePattern[" + stmt + "] not supported"); + } + + Map.Entry<RdfCloudTripleStoreConstants.TABLE_LAYOUT, ByteRange> entry = + strategy.defineRange(stmt.getSubject(), stmt.getPredicate(), stmt.getObject(), stmt.getContext(), conf); + + //use range to set scanner + //populate scanner based on authorizations, ttl + layout = entry.getKey(); + ByteRange byteRange = entry.getValue(); + Range range = new Range(new Text(byteRange.getStart()), new Text(byteRange.getEnd())); + ranges.add(range); + rangeMap.ranges.add(new RdfCloudTripleStoreUtils.CustomEntry<Range, BindingSet>(range, bs)); + } + //no ranges + if (layout == null) return null; + String regexSubject = conf.getRegexSubject(); + String regexPredicate = conf.getRegexPredicate(); + String regexObject = conf.getRegexObject(); + TripleRowRegex tripleRowRegex = strategy.buildRegex(regexSubject, regexPredicate, regexObject, null, null); + + String table = layoutToTable(layout, conf); + boolean useBatchScanner = ranges.size() > maxRanges; + RyaStatementBindingSetKeyValueIterator iterator = null; + if (useBatchScanner) { + ScannerBase scanner = connector.createBatchScanner(table, authorizations, numThreads); + ((BatchScanner) scanner).setRanges(ranges); + fillScanner(scanner, context, null, ttl, null, tripleRowRegex, conf); + iterator = new RyaStatementBindingSetKeyValueIterator(layout, ryaContext, scanner, rangeMap); + } else { + Scanner scannerBase = null; + Iterator<Map.Entry<Key, Value>>[] iters = new Iterator[ranges.size()]; + int i = 0; + for (Range range : ranges) { + scannerBase = connector.createScanner(table, authorizations); + scannerBase.setRange(range); + fillScanner(scannerBase, context, null, ttl, null, tripleRowRegex, conf); + iters[i] = scannerBase.iterator(); + i++; + } + iterator = new RyaStatementBindingSetKeyValueIterator(layout, Iterators.concat(iters), rangeMap, ryaContext); + } + if (maxResults != null) { + iterator.setMaxResults(maxResults); + } + return iterator; + } catch (Exception e) { + throw new RyaDAOException(e); + } + + } + + @Override + public CloseableIteration<RyaStatement, RyaDAOException> batchQuery(Collection<RyaStatement> stmts, AccumuloRdfConfiguration conf) + throws RyaDAOException { + if (conf == null) { + conf = configuration; + } + + BatchRyaQuery batchRyaQuery = BatchRyaQuery.builder(stmts).load(conf).build(); + CloseableIterable<RyaStatement> results = query(batchRyaQuery); + + return new CloseableIterableIteration<RyaStatement, RyaDAOException>(results); + } + + @Override + public CloseableIterable<RyaStatement> query(RyaQuery ryaQuery) throws RyaDAOException { + Preconditions.checkNotNull(ryaQuery); + RyaStatement stmt = ryaQuery.getQuery(); + Preconditions.checkNotNull(stmt); + + //query configuration + String[] auths = ryaQuery.getAuths(); + Authorizations authorizations = auths != null ? new Authorizations(auths) : configuration.getAuthorizations(); + Long ttl = ryaQuery.getTtl(); + Long currentTime = ryaQuery.getCurrentTime(); + Long maxResults = ryaQuery.getMaxResults(); + Integer batchSize = ryaQuery.getBatchSize(); + String regexSubject = ryaQuery.getRegexSubject(); + String regexPredicate = ryaQuery.getRegexPredicate(); + String regexObject = ryaQuery.getRegexObject(); + TableLayoutStrategy tableLayoutStrategy = configuration.getTableLayoutStrategy(); + + try { + //find triple pattern range + TriplePatternStrategy strategy = ryaContext.retrieveStrategy(stmt); + TABLE_LAYOUT layout; + Range range; + RyaURI subject = stmt.getSubject(); + RyaURI predicate = stmt.getPredicate(); + RyaType object = stmt.getObject(); + RyaURI context = stmt.getContext(); + String qualifier = stmt.getQualifer(); + TripleRowRegex tripleRowRegex = null; + if (strategy != null) { + //otherwise, full table scan is supported + Map.Entry<RdfCloudTripleStoreConstants.TABLE_LAYOUT, ByteRange> entry = + strategy.defineRange(subject, predicate, object, context, null); + layout = entry.getKey(); + ByteRange byteRange = entry.getValue(); + range = new Range(new Text(byteRange.getStart()), new Text(byteRange.getEnd())); + + } else { + range = new Range(); + layout = TABLE_LAYOUT.SPO; + strategy = ryaContext.retrieveStrategy(layout); + } + + byte[] objectTypeInfo = null; + if (object != null) { + //TODO: Not good to serialize this twice + if (object instanceof RyaRange) { + objectTypeInfo = RyaContext.getInstance().serializeType(((RyaRange) object).getStart())[1]; + } else { + objectTypeInfo = RyaContext.getInstance().serializeType(object)[1]; + } + } + + tripleRowRegex = strategy.buildRegex(regexSubject, regexPredicate, regexObject, null, objectTypeInfo); + + //use range to set scanner + //populate scanner based on authorizations, ttl + String table = layoutToTable(layout, tableLayoutStrategy); + Scanner scanner = connector.createScanner(table, authorizations); + scanner.setRange(range); + if (batchSize != null) { + scanner.setBatchSize(batchSize); + } + fillScanner(scanner, context, qualifier, ttl, currentTime, tripleRowRegex, ryaQuery.getConf()); + + FluentCloseableIterable<RyaStatement> results = FluentCloseableIterable.from(new ScannerBaseCloseableIterable(scanner)) + .transform(keyValueToRyaStatementFunctionMap.get(layout)); + if (maxResults != null) { + results = results.limit(maxResults.intValue()); + } + + return results; + } catch (Exception e) { + throw new RyaDAOException(e); + } + } + + @Override + public CloseableIterable<RyaStatement> query(BatchRyaQuery ryaQuery) throws RyaDAOException { + Preconditions.checkNotNull(ryaQuery); + Iterable<RyaStatement> stmts = ryaQuery.getQueries(); + Preconditions.checkNotNull(stmts); + + //query configuration + String[] auths = ryaQuery.getAuths(); + final Authorizations authorizations = auths != null ? new Authorizations(auths) : configuration.getAuthorizations(); + final Long ttl = ryaQuery.getTtl(); + Long currentTime = ryaQuery.getCurrentTime(); + Long maxResults = ryaQuery.getMaxResults(); + Integer batchSize = ryaQuery.getBatchSize(); + Integer numQueryThreads = ryaQuery.getNumQueryThreads(); + String regexSubject = ryaQuery.getRegexSubject(); + String regexPredicate = ryaQuery.getRegexPredicate(); + String regexObject = ryaQuery.getRegexObject(); + TableLayoutStrategy tableLayoutStrategy = configuration.getTableLayoutStrategy(); + int maxRanges = ryaQuery.getMaxRanges(); + + //TODO: cannot span multiple tables here + try { + Collection<Range> ranges = new HashSet<Range>(); + TABLE_LAYOUT layout = null; + RyaURI context = null; + TriplePatternStrategy strategy = null; + for (RyaStatement stmt : stmts) { + context = stmt.getContext(); //TODO: This will be overwritten + strategy = ryaContext.retrieveStrategy(stmt); + if (strategy == null) { + throw new IllegalArgumentException("TriplePattern[" + stmt + "] not supported"); + } + + Map.Entry<RdfCloudTripleStoreConstants.TABLE_LAYOUT, ByteRange> entry = + strategy.defineRange(stmt.getSubject(), stmt.getPredicate(), stmt.getObject(), stmt.getContext(), null); + + //use range to set scanner + //populate scanner based on authorizations, ttl + layout = entry.getKey(); + ByteRange byteRange = entry.getValue(); + Range range = new Range(new Text(byteRange.getStart()), new Text(byteRange.getEnd())); + ranges.add(range); + } + //no ranges + if (layout == null) throw new IllegalArgumentException("No table layout specified"); + + final TripleRowRegex tripleRowRegex = strategy.buildRegex(regexSubject, regexPredicate, regexObject, null, null); + + final String table = layoutToTable(layout, tableLayoutStrategy); + boolean useBatchScanner = ranges.size() > maxRanges; + FluentCloseableIterable<RyaStatement> results = null; + if (useBatchScanner) { + BatchScanner scanner = connector.createBatchScanner(table, authorizations, numQueryThreads); + scanner.setRanges(ranges); + fillScanner(scanner, context, null, ttl, null, tripleRowRegex, ryaQuery.getConf()); + results = FluentCloseableIterable.from(new ScannerBaseCloseableIterable(scanner)).transform(keyValueToRyaStatementFunctionMap.get(layout)); + } else { + final RyaURI fcontext = context; + final RdfCloudTripleStoreConfiguration fconf = ryaQuery.getConf(); + FluentIterable<RyaStatement> fluent = FluentIterable.from(ranges).transformAndConcat(new Function<Range, Iterable<Map.Entry<Key, Value>>>() { + @Override + public Iterable<Map.Entry<Key, Value>> apply(Range range) { + try { + Scanner scanner = connector.createScanner(table, authorizations); + scanner.setRange(range); + fillScanner(scanner, fcontext, null, ttl, null, tripleRowRegex, fconf); + return scanner; + } catch (Exception e) { + throw new RuntimeException(e); + } + } + }).transform(keyValueToRyaStatementFunctionMap.get(layout)); + + results = FluentCloseableIterable.from(CloseableIterables.wrap(fluent)); + } + if (maxResults != null) { + results = results.limit(maxResults.intValue()); + } + return results; + } catch (Exception e) { + throw new RyaDAOException(e); + } + } + + protected void fillScanner(ScannerBase scanner, RyaURI context, String qualifier, Long ttl, Long currentTime, TripleRowRegex tripleRowRegex, RdfCloudTripleStoreConfiguration conf) throws IOException { + if (context != null && qualifier != null) { + scanner.fetchColumn(new Text(context.getData()), new Text(qualifier)); + } else if (context != null) { + scanner.fetchColumnFamily(new Text(context.getData())); + } else if (qualifier != null) { + IteratorSetting setting = new IteratorSetting(8, "riq", RegExFilter.class.getName()); + RegExFilter.setRegexs(setting, null, null, qualifier, null, false); + scanner.addScanIterator(setting); + } + if (ttl != null) { + IteratorSetting setting = new IteratorSetting(9, "fi", TimestampFilter.class.getName()); + TimestampFilter.setStart(setting, System.currentTimeMillis() - ttl, true); + if(currentTime != null){ + TimestampFilter.setStart(setting, currentTime - ttl, true); + TimestampFilter.setEnd(setting, currentTime, true); + } + scanner.addScanIterator(setting); + } + if (tripleRowRegex != null) { + IteratorSetting setting = new IteratorSetting(11, "ri", RegExFilter.class.getName()); + String regex = tripleRowRegex.getRow(); + RegExFilter.setRegexs(setting, regex, null, null, null, false); + scanner.addScanIterator(setting); + } + if (conf instanceof AccumuloRdfConfiguration) { + //TODO should we take the iterator settings as is or should we adjust the priority based on the above? + for (IteratorSetting itr : ((AccumuloRdfConfiguration)conf).getAdditionalIterators()) { + scanner.addScanIterator(itr); + } + } + } + + @Override + public void setConf(AccumuloRdfConfiguration conf) { + this.configuration = conf; + } + + @Override + public AccumuloRdfConfiguration getConf() { + return configuration; + } +}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/44a2dcf0/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/KeyValueToRyaStatementFunction.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/KeyValueToRyaStatementFunction.java b/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/KeyValueToRyaStatementFunction.java new file mode 100644 index 0000000..2813438 --- /dev/null +++ b/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/KeyValueToRyaStatementFunction.java @@ -0,0 +1,72 @@ +package mvm.rya.accumulo.query; + +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + */ + + + +import java.util.Map; + +import mvm.rya.api.RdfCloudTripleStoreConfiguration; +import mvm.rya.api.RdfCloudTripleStoreConstants.TABLE_LAYOUT; +import mvm.rya.api.domain.RyaStatement; +import mvm.rya.api.resolver.RyaTripleContext; +import mvm.rya.api.resolver.triple.TripleRow; +import mvm.rya.api.resolver.triple.TripleRowResolverException; + +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.Value; + +import com.google.common.base.Function; + +/** + * Date: 1/30/13 + * Time: 2:09 PM + */ +public class KeyValueToRyaStatementFunction implements Function<Map.Entry<Key, Value>, RyaStatement> { + + private TABLE_LAYOUT tableLayout; + private RyaTripleContext context; + + public KeyValueToRyaStatementFunction(TABLE_LAYOUT tableLayout, RyaTripleContext context) { + this.tableLayout = tableLayout; + this.context = context; + } + + @Override + public RyaStatement apply(Map.Entry<Key, Value> input) { + Key key = input.getKey(); + Value value = input.getValue(); + RyaStatement statement = null; + try { + statement = context.deserializeTriple(tableLayout, + new TripleRow(key.getRowData().toArray(), + key.getColumnFamilyData().toArray(), + key.getColumnQualifierData().toArray(), + key.getTimestamp(), + key.getColumnVisibilityData().toArray(), + (value != null) ? value.get() : null + )); + } catch (TripleRowResolverException e) { + throw new RuntimeException(e); + } + + return statement; + } +} http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/44a2dcf0/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/RangeBindingSetEntries.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/RangeBindingSetEntries.java b/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/RangeBindingSetEntries.java new file mode 100644 index 0000000..c59cb87 --- /dev/null +++ b/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/RangeBindingSetEntries.java @@ -0,0 +1,58 @@ +package mvm.rya.accumulo.query; + +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + */ + + + +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.Range; +import org.openrdf.query.BindingSet; + +import java.util.ArrayList; +import java.util.Collection; +import java.util.Map; + +/** + * Class RangeBindingSetCollection + * Date: Feb 23, 2011 + * Time: 10:15:48 AM + */ +public class RangeBindingSetEntries { + public Collection<Map.Entry<Range, BindingSet>> ranges; + + public RangeBindingSetEntries() { + this(new ArrayList<Map.Entry<Range, BindingSet>>()); + } + + public RangeBindingSetEntries(Collection<Map.Entry<Range, BindingSet>> ranges) { + this.ranges = ranges; + } + + public Collection<BindingSet> containsKey(Key key) { + //TODO: need to find a better way to sort these and pull + //TODO: maybe fork/join here + Collection<BindingSet> bss = new ArrayList<BindingSet>(); + for (Map.Entry<Range, BindingSet> entry : ranges) { + if (entry.getKey().contains(key)) + bss.add(entry.getValue()); + } + return bss; + } +} http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/44a2dcf0/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/RyaStatementBindingSetKeyValueIterator.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/RyaStatementBindingSetKeyValueIterator.java b/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/RyaStatementBindingSetKeyValueIterator.java new file mode 100644 index 0000000..b4333bd --- /dev/null +++ b/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/RyaStatementBindingSetKeyValueIterator.java @@ -0,0 +1,154 @@ +package mvm.rya.accumulo.query; + +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + */ + + + +import info.aduna.iteration.CloseableIteration; + +import java.util.Collection; +import java.util.Iterator; +import java.util.Map; +import java.util.NoSuchElementException; + +import mvm.rya.api.RdfCloudTripleStoreConstants.TABLE_LAYOUT; +import mvm.rya.api.RdfCloudTripleStoreUtils; +import mvm.rya.api.domain.RyaStatement; +import mvm.rya.api.persist.RyaDAOException; +import mvm.rya.api.resolver.RyaContext; +import mvm.rya.api.resolver.RyaTripleContext; +import mvm.rya.api.resolver.triple.TripleRow; +import mvm.rya.api.resolver.triple.TripleRowResolverException; + +import org.apache.accumulo.core.client.BatchScanner; +import org.apache.accumulo.core.client.Scanner; +import org.apache.accumulo.core.client.ScannerBase; +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.Value; +import org.openrdf.query.BindingSet; + +/** + * Date: 7/17/12 + * Time: 11:48 AM + */ +public class RyaStatementBindingSetKeyValueIterator implements CloseableIteration<Map.Entry<RyaStatement, BindingSet>, RyaDAOException> { + private Iterator<Map.Entry<Key, Value>> dataIterator; + private TABLE_LAYOUT tableLayout; + private Long maxResults = -1L; + private ScannerBase scanner; + private boolean isBatchScanner; + private RangeBindingSetEntries rangeMap; + private Iterator<BindingSet> bsIter; + private RyaStatement statement; + private RyaTripleContext ryaContext; + + public RyaStatementBindingSetKeyValueIterator(TABLE_LAYOUT tableLayout, RyaTripleContext context, ScannerBase scannerBase, RangeBindingSetEntries rangeMap) { + this(tableLayout, ((scannerBase instanceof BatchScanner) ? ((BatchScanner) scannerBase).iterator() : ((Scanner) scannerBase).iterator()), rangeMap, context); + this.scanner = scannerBase; + isBatchScanner = scanner instanceof BatchScanner; + } + + public RyaStatementBindingSetKeyValueIterator(TABLE_LAYOUT tableLayout, Iterator<Map.Entry<Key, Value>> dataIterator, RangeBindingSetEntries rangeMap, RyaTripleContext ryaContext) { + this.tableLayout = tableLayout; + this.rangeMap = rangeMap; + this.dataIterator = dataIterator; + this.ryaContext = ryaContext; + } + + @Override + public void close() throws RyaDAOException { + dataIterator = null; + if (scanner != null && isBatchScanner) { + ((BatchScanner) scanner).close(); + } + } + + public boolean isClosed() throws RyaDAOException { + return dataIterator == null; + } + + @Override + public boolean hasNext() throws RyaDAOException { + if (isClosed()) { + return false; + } + if (maxResults != 0) { + if (bsIter != null && bsIter.hasNext()) { + return true; + } + if (dataIterator.hasNext()) { + return true; + } else { + maxResults = 0l; + return false; + } + } + return false; + } + + @Override + public Map.Entry<RyaStatement, BindingSet> next() throws RyaDAOException { + if (!hasNext() || isClosed()) { + throw new NoSuchElementException(); + } + + try { + while (true) { + if (bsIter != null && bsIter.hasNext()) { + maxResults--; + return new RdfCloudTripleStoreUtils.CustomEntry<RyaStatement, BindingSet>(statement, bsIter.next()); + } + + if (dataIterator.hasNext()) { + Map.Entry<Key, Value> next = dataIterator.next(); + Key key = next.getKey(); + statement = ryaContext.deserializeTriple(tableLayout, + new TripleRow(key.getRowData().toArray(), key.getColumnFamilyData().toArray(), key.getColumnQualifierData().toArray(), + key.getTimestamp(), key.getColumnVisibilityData().toArray(), next.getValue().get())); + if (next.getValue() != null) { + statement.setValue(next.getValue().get()); + } + Collection<BindingSet> bindingSets = rangeMap.containsKey(key); + if (!bindingSets.isEmpty()) { + bsIter = bindingSets.iterator(); + } + } else { + break; + } + } + return null; + } catch (TripleRowResolverException e) { + throw new RyaDAOException(e); + } + } + + @Override + public void remove() throws RyaDAOException { + next(); + } + + public Long getMaxResults() { + return maxResults; + } + + public void setMaxResults(Long maxResults) { + this.maxResults = maxResults; + } +} http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/44a2dcf0/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/RyaStatementKeyValueIterator.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/RyaStatementKeyValueIterator.java b/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/RyaStatementKeyValueIterator.java new file mode 100644 index 0000000..f4c3081 --- /dev/null +++ b/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/RyaStatementKeyValueIterator.java @@ -0,0 +1,107 @@ +package mvm.rya.accumulo.query; + +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + */ + + + +import info.aduna.iteration.CloseableIteration; + +import java.util.Iterator; +import java.util.Map; + +import mvm.rya.api.RdfCloudTripleStoreConstants.TABLE_LAYOUT; +import mvm.rya.api.domain.RyaStatement; +import mvm.rya.api.persist.RyaDAOException; +import mvm.rya.api.resolver.RyaContext; +import mvm.rya.api.resolver.RyaTripleContext; +import mvm.rya.api.resolver.triple.TripleRow; +import mvm.rya.api.resolver.triple.TripleRowResolverException; + +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.Value; + +/** + * Date: 7/17/12 + * Time: 11:48 AM + */ +public class RyaStatementKeyValueIterator implements CloseableIteration<RyaStatement, RyaDAOException> { + private Iterator<Map.Entry<Key, Value>> dataIterator; + private TABLE_LAYOUT tableLayout; + private Long maxResults = -1L; + private RyaTripleContext context; + + public RyaStatementKeyValueIterator(TABLE_LAYOUT tableLayout, RyaTripleContext context, Iterator<Map.Entry<Key, Value>> dataIterator) { + this.tableLayout = tableLayout; + this.dataIterator = dataIterator; + this.context = context; + } + + @Override + public void close() throws RyaDAOException { + dataIterator = null; + } + + public boolean isClosed() throws RyaDAOException { + return dataIterator == null; + } + + @Override + public boolean hasNext() throws RyaDAOException { + if (isClosed()) { + throw new RyaDAOException("Closed Iterator"); + } + return maxResults != 0 && dataIterator.hasNext(); + } + + @Override + public RyaStatement next() throws RyaDAOException { + if (!hasNext()) { + return null; + } + + try { + Map.Entry<Key, Value> next = dataIterator.next(); + Key key = next.getKey(); + RyaStatement statement = context.deserializeTriple(tableLayout, + new TripleRow(key.getRowData().toArray(), key.getColumnFamilyData().toArray(), key.getColumnQualifierData().toArray(), + key.getTimestamp(), key.getColumnVisibilityData().toArray(), next.getValue().get())); + if (next.getValue() != null) { + statement.setValue(next.getValue().get()); + } + maxResults--; + return statement; + } catch (TripleRowResolverException e) { + throw new RyaDAOException(e); + } + } + + @Override + public void remove() throws RyaDAOException { + next(); + } + + public Long getMaxResults() { + return maxResults; + } + + public void setMaxResults(Long maxResults) { + this.maxResults = maxResults; + } +} http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/44a2dcf0/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/ScannerBaseCloseableIterable.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/ScannerBaseCloseableIterable.java b/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/ScannerBaseCloseableIterable.java new file mode 100644 index 0000000..d2dcef9 --- /dev/null +++ b/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/query/ScannerBaseCloseableIterable.java @@ -0,0 +1,56 @@ +package mvm.rya.accumulo.query; + +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + */ + + + +import com.google.common.base.Preconditions; +import org.apache.accumulo.core.client.ScannerBase; +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.Value; +import org.calrissian.mango.collect.AbstractCloseableIterable; + +import java.io.IOException; +import java.util.Iterator; +import java.util.Map; + +/** + * Date: 1/30/13 + * Time: 2:15 PM + */ +public class ScannerBaseCloseableIterable extends AbstractCloseableIterable<Map.Entry<Key, Value>> { + + protected ScannerBase scanner; + + public ScannerBaseCloseableIterable(ScannerBase scanner) { + Preconditions.checkNotNull(scanner); + this.scanner = scanner; + } + + @Override + protected void doClose() throws IOException { + scanner.close(); + } + + @Override + protected Iterator<Map.Entry<Key, Value>> retrieveIterator() { + return scanner.iterator(); + } +} http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/44a2dcf0/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/utils/TimeRangeFilter.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/utils/TimeRangeFilter.java b/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/utils/TimeRangeFilter.java new file mode 100644 index 0000000..97d2f54 --- /dev/null +++ b/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/utils/TimeRangeFilter.java @@ -0,0 +1,87 @@ +package mvm.rya.accumulo.utils; + +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + */ + + + +import org.apache.accumulo.core.data.Key; +import org.apache.accumulo.core.data.Value; +import org.apache.accumulo.core.iterators.Filter; +import org.apache.accumulo.core.iterators.IteratorEnvironment; +import org.apache.accumulo.core.iterators.OptionDescriber; +import org.apache.accumulo.core.iterators.SortedKeyValueIterator; + +import java.io.IOException; +import java.util.Map; +import java.util.TreeMap; + +/** + * Set the startTime and timeRange. The filter will only keyValues that + * are within the range [startTime - timeRange, startTime]. + */ +public class TimeRangeFilter extends Filter { + private long timeRange; + private long startTime; + public static final String TIME_RANGE_PROP = "timeRange"; + public static final String START_TIME_PROP = "startTime"; + + @Override + public boolean accept(Key k, Value v) { + long diff = startTime - k.getTimestamp(); + return !(diff > timeRange || diff < 0); + } + + @Override + public void init(SortedKeyValueIterator<Key,Value> source, Map<String,String> options, IteratorEnvironment env) throws IOException { + super.init(source, options, env); + if (options == null) { + throw new IllegalArgumentException("options must be set for TimeRangeFilter"); + } + + timeRange = -1; + String timeRange_s = options.get(TIME_RANGE_PROP); + if (timeRange_s == null) + throw new IllegalArgumentException("timeRange must be set for TimeRangeFilter"); + + timeRange = Long.parseLong(timeRange_s); + + String time = options.get(START_TIME_PROP); + if (time != null) + startTime = Long.parseLong(time); + else + startTime = System.currentTimeMillis(); + } + + @Override + public OptionDescriber.IteratorOptions describeOptions() { + Map<String, String> options = new TreeMap<String, String>(); + options.put(TIME_RANGE_PROP, "time range from the startTime (milliseconds)"); + options.put(START_TIME_PROP, "if set, use the given value as the absolute time in milliseconds as the start time in the time range."); + return new OptionDescriber.IteratorOptions("timeRangeFilter", "TimeRangeFilter removes entries with timestamps outside of the given time range: " + + "[startTime - timeRange, startTime]", + options, null); + } + + @Override + public boolean validateOptions(Map<String, String> options) { + Long.parseLong(options.get(TIME_RANGE_PROP)); + return true; + } +} http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/44a2dcf0/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/utils/VisibilitySimplifier.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/utils/VisibilitySimplifier.java b/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/utils/VisibilitySimplifier.java new file mode 100644 index 0000000..cc4edca --- /dev/null +++ b/dao/accumulo.rya/src/main/java/org/apache/rya/accumulo/utils/VisibilitySimplifier.java @@ -0,0 +1,54 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you 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. + */ +package mvm.rya.accumulo.utils; + +import static java.util.Objects.requireNonNull; + +import javax.annotation.ParametersAreNonnullByDefault; + +import org.apache.accumulo.core.security.ColumnVisibility; + +import com.google.common.base.Charsets; + +/** + * Simplifies Accumulo visibility expressions. + */ +@ParametersAreNonnullByDefault +public class VisibilitySimplifier { + + /** + * Simplifies an Accumulo visibility expression. + * + * @param visibility - The expression to simplify. (not null) + * @return A simplified form of {@code visibility}. + */ + public String simplify(final String visibility) { + requireNonNull(visibility); + + String last = visibility; + String simplified = new String(new ColumnVisibility(visibility).flatten(), Charsets.UTF_8); + + while(!simplified.equals(last)) { + last = simplified; + simplified = new String(new ColumnVisibility(simplified).flatten(), Charsets.UTF_8); + } + + return simplified; + } +} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/44a2dcf0/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/AccumuloITBase.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/AccumuloITBase.java b/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/AccumuloITBase.java deleted file mode 100644 index 7dd23e6..0000000 --- a/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/AccumuloITBase.java +++ /dev/null @@ -1,105 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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. - */ -package mvm.rya.accumulo; - -import java.io.IOException; - -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.minicluster.MiniAccumuloCluster; -import org.apache.log4j.Level; -import org.apache.log4j.Logger; -import org.apache.zookeeper.ClientCnxn; -import org.junit.After; -import org.junit.Before; -import org.junit.BeforeClass; - -/** - * Boilerplate code for a unit test that uses a {@link MiniAccumuloCluster}. - * <p> - * It uses the same instance of {@link MiniAccumuloCluster} and just clears out - * any tables that were added between tests. - */ -public class AccumuloITBase { - - // Managed the MiniAccumuloCluster - private MiniAccumuloClusterInstance cluster = null; - - @BeforeClass - public static void killLoudLogs() { - Logger.getLogger(ClientCnxn.class).setLevel(Level.ERROR); - } - - @Before - public void initCluster() throws IOException, InterruptedException, AccumuloException, AccumuloSecurityException { - cluster = new MiniAccumuloClusterInstance(); - cluster.startMiniAccumulo(); - } - - - @After - public void tearDownCluster() throws IOException, InterruptedException { - cluster.stopMiniAccumulo(); - } - - /** - * @return The {@link MiniAccumuloClusterInstance} used by the tests. - */ - public MiniAccumuloClusterInstance getClusterInstance() { - return cluster; - } - - /** - * @return The root username. - */ - public String getUsername() { - return cluster.getUsername(); - } - - /** - * @return The root password. - */ - public String getPassword() { - return cluster.getPassword(); - } - - /** - * @return The MiniAccumulo's zookeeper instance name. - */ - public String getInstanceName() { - return cluster.getInstanceName(); - } - - /** - * @return The MiniAccumulo's zookeepers. - */ - public String getZookeepers() { - return cluster.getZookeepers(); - } - - /** - * @return A {@link Connector} that creates connections to the mini accumulo cluster. - * @throws AccumuloException Could not connect to the cluster. - * @throws AccumuloSecurityException Could not connect to the cluster because of a security violation. - */ - public Connector getConnector() throws AccumuloException, AccumuloSecurityException { - return cluster.getConnector(); - } -} \ No newline at end of file http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/44a2dcf0/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/AccumuloRdfConfigurationTest.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/AccumuloRdfConfigurationTest.java b/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/AccumuloRdfConfigurationTest.java deleted file mode 100644 index ffd316e..0000000 --- a/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/AccumuloRdfConfigurationTest.java +++ /dev/null @@ -1,75 +0,0 @@ -package mvm.rya.accumulo; - -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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. - */ - - - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertTrue; - -import java.util.Arrays; -import java.util.HashMap; -import java.util.Map; - -import org.apache.accumulo.core.client.IteratorSetting; -import org.apache.accumulo.core.security.Authorizations; -import org.junit.Test; -import org.slf4j.Logger; -import org.slf4j.LoggerFactory; - -public class AccumuloRdfConfigurationTest { - private static final Logger logger = LoggerFactory.getLogger(AccumuloRdfConfigurationTest.class); - - @Test - public void testAuths() { - String[] arr = {"U", "FOUO"}; - String str = "U,FOUO"; - Authorizations auths = new Authorizations(arr); - - AccumuloRdfConfiguration conf = new AccumuloRdfConfiguration(); - - conf.setAuths(arr); - assertTrue(Arrays.equals(arr, conf.getAuths())); - assertEquals(str, conf.getAuth()); - assertEquals(auths, conf.getAuthorizations()); - - conf.setAuth(str); - assertTrue(Arrays.equals(arr, conf.getAuths())); - assertEquals(str, conf.getAuth()); - assertEquals(auths, conf.getAuthorizations()); - } - - @Test - public void testIterators() { - AccumuloRdfConfiguration conf = new AccumuloRdfConfiguration(); - - Map<String, String> options = new HashMap<String, String>(); - options.put("key1", "value1"); - options.put("key2", "value2"); - IteratorSetting setting = new IteratorSetting(1, "test", "test2", options); - - conf.setAdditionalIterators(setting); - IteratorSetting[] iteratorSettings = conf.getAdditionalIterators(); - assertTrue(iteratorSettings.length == 1); - - assertEquals(setting, iteratorSettings[0]); - - } -} http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/44a2dcf0/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/AccumuloRyaDAOTest.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/AccumuloRyaDAOTest.java b/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/AccumuloRyaDAOTest.java deleted file mode 100644 index 8b55070..0000000 --- a/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/AccumuloRyaDAOTest.java +++ /dev/null @@ -1,713 +0,0 @@ -package mvm.rya.accumulo; - -/* - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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. - */ - - - -import static org.junit.Assert.assertEquals; -import static org.junit.Assert.assertFalse; -import static org.junit.Assert.assertNull; -import static org.junit.Assert.assertTrue; -import info.aduna.iteration.CloseableIteration; - -import java.util.ArrayList; -import java.util.Collection; -import java.util.List; -import java.util.UUID; - -import mvm.rya.accumulo.query.AccumuloRyaQueryEngine; -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.persist.query.RyaQuery; -import mvm.rya.api.resolver.RdfToRyaConversions; -import mvm.rya.api.resolver.RyaContext; - -import org.apache.accumulo.core.client.Connector; -import org.apache.accumulo.core.client.IteratorSetting; -import org.apache.accumulo.core.client.Scanner; -import org.apache.accumulo.core.client.TableNotFoundException; -import org.apache.accumulo.core.client.mock.MockInstance; -import org.apache.accumulo.core.iterators.FirstEntryInRowIterator; -import org.calrissian.mango.collect.FluentCloseableIterable; -import org.junit.After; -import org.junit.Before; -import org.junit.Test; -import org.openrdf.model.ValueFactory; -import org.openrdf.model.impl.ValueFactoryImpl; -import org.openrdf.model.vocabulary.XMLSchema; - -/** - * Class AccumuloRdfDAOTest - * Date: Mar 7, 2012 - * Time: 9:42:28 AM - */ -public class AccumuloRyaDAOTest { - - private AccumuloRyaDAO dao; - private ValueFactory vf = new ValueFactoryImpl(); - static String litdupsNS = "urn:test:litdups#"; - private AccumuloRdfConfiguration conf; - private Connector connector; - - @Before - public void setUp() throws Exception { - dao = new AccumuloRyaDAO(); - connector = new MockInstance().getConnector("", ""); - dao.setConnector(connector); - conf = new AccumuloRdfConfiguration(); - dao.setConf(conf); - dao.init(); - } - - @After - public void tearDown() throws Exception { - dao.purge(conf); - dao.destroy(); - } - - @Test - public void testAdd() throws Exception { - RyaURI cpu = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "cpu")); - RyaURI loadPerc = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "loadPerc")); - RyaURI uri1 = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "uri1")); - dao.add(new RyaStatement(cpu, loadPerc, uri1)); - - CloseableIteration<RyaStatement, RyaDAOException> iter = dao.getQueryEngine().query(new RyaStatement(cpu, loadPerc, null), conf); - int count = 0; - while (iter.hasNext()) { - assertTrue(uri1.equals(iter.next().getObject())); - count++; - } - iter.close(); - assertEquals(1, count); - - dao.delete(new RyaStatement(cpu, loadPerc, null), conf); - - iter = dao.getQueryEngine().query(new RyaStatement(cpu, loadPerc, null), conf); - count = 0; - while (iter.hasNext()) { - count++; - iter.next(); - } - iter.close(); - assertEquals(0, count); - } - - @Test - public void testDeleteDiffVisibility() throws Exception { - RyaURI cpu = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "cpu")); - RyaURI loadPerc = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "loadPerc")); - RyaURI uri1 = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "uri1")); - RyaStatement stmt1 = new RyaStatement(cpu, loadPerc, uri1, null, "1", "vis1".getBytes()); - dao.add(stmt1); - RyaStatement stmt2 = new RyaStatement(cpu, loadPerc, uri1, null, "2", "vis2".getBytes()); - dao.add(stmt2); - - AccumuloRdfConfiguration cloneConf = conf.clone(); - cloneConf.setAuth("vis1,vis2"); - - CloseableIteration<RyaStatement, RyaDAOException> iter = dao.getQueryEngine().query(new RyaStatement(cpu, loadPerc, null), cloneConf); - int count = 0; - while (iter.hasNext()) { - iter.next(); - count++; - } - iter.close(); - assertEquals(2, count); - - dao.delete(stmt1, cloneConf); - - iter = dao.getQueryEngine().query(new RyaStatement(cpu, loadPerc, null), cloneConf); - count = 0; - while (iter.hasNext()) { - iter.next(); - count++; - } - iter.close(); - assertEquals(1, count); - } - - @Test - public void testDeleteDiffTimestamp() throws Exception { - RyaURI cpu = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "cpu")); - RyaURI loadPerc = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "loadPerc")); - RyaURI uri1 = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "uri1")); - RyaStatement stmt1 = new RyaStatement(cpu, loadPerc, uri1, null, "1", null, null, 100l); - dao.add(stmt1); - RyaStatement stmt2 = new RyaStatement(cpu, loadPerc, uri1, null, "2", null, null, 100l); - dao.add(stmt2); - - int resultSize = FluentCloseableIterable.from(dao.getQueryEngine().query( - RyaQuery.builder(new RyaStatement(cpu, loadPerc, null)).build())).autoClose().size(); - assertEquals(2, resultSize); - - final RyaStatement addStmt = new RyaStatement(cpu, loadPerc, uri1, null, "1", - null, null, 101l); - dao.delete(stmt1, conf); - dao.add(addStmt); - - resultSize = FluentCloseableIterable.from(dao.getQueryEngine().query( - RyaQuery.builder(new RyaStatement(cpu, loadPerc, null)).build())).autoClose().size(); - assertEquals(2, resultSize); //the delete marker should not delete the new stmt - } - - @Test - public void testDelete() throws Exception { - RyaURI predicate = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "pred")); - RyaURI subj = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "subj")); - - // create a "bulk load" of 10,000 statements - int statement_count = 10000; - for (int i = 0 ; i < statement_count ; i++){ - //make the statement very large so we will get a lot of random flushes - RyaURI obj = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, String.format("object%050d",i))); - RyaStatement stmt = new RyaStatement(subj, predicate, obj); - dao.add(stmt); - } - - CloseableIteration<RyaStatement, RyaDAOException> iter; - - //check to see if all of the statements made it to the subj table - //delete based on the data in the subj table - RyaStatement subjQuery = new RyaStatement(subj, null, null); - iter = dao.getQueryEngine().query(subjQuery, conf); - List<RyaStatement> stmts = new ArrayList<RyaStatement>(); - while (iter.hasNext()) { - stmts.add(iter.next()); - } - assertEquals(statement_count, stmts.size()); - dao.delete(stmts.iterator(), conf); - - // check statements in the predicate table - RyaStatement predQuery = new RyaStatement(null, predicate, null); - iter = dao.getQueryEngine().query(predQuery, conf); - int count = 0; - while (iter.hasNext()) { - count++; - } - iter.close(); - assertEquals(0, count); - } - - @Test - public void testAddEmptyString() throws Exception { - RyaURI cpu = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "cpu")); - RyaURI loadPerc = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "loadPerc")); - RyaType empty = new RyaType(""); - dao.add(new RyaStatement(cpu, loadPerc, empty)); - - CloseableIteration<RyaStatement, RyaDAOException> iter = dao.getQueryEngine().query(new RyaStatement(cpu, loadPerc, null), conf); - while (iter.hasNext()) { - assertEquals("", iter.next().getObject().getData()); - } - iter.close(); - } - - @Test - public void testMaxResults() throws Exception { - RyaURI cpu = new RyaURI(litdupsNS + "cpu"); - RyaURI loadPerc = new RyaURI(litdupsNS + "loadPerc"); - dao.add(new RyaStatement(cpu, loadPerc, new RyaURI(litdupsNS + "uri1"))); - dao.add(new RyaStatement(cpu, loadPerc, new RyaURI(litdupsNS + "uri2"))); - dao.add(new RyaStatement(cpu, loadPerc, new RyaURI(litdupsNS + "uri3"))); - dao.add(new RyaStatement(cpu, loadPerc, new RyaURI(litdupsNS + "uri4"))); - dao.add(new RyaStatement(cpu, loadPerc, new RyaURI(litdupsNS + "uri5"))); - - AccumuloRyaQueryEngine queryEngine = dao.getQueryEngine(); - AccumuloRdfConfiguration queryConf = new AccumuloRdfConfiguration(conf); - long limit = 3l; - queryConf.setLimit(limit); - - CloseableIteration<RyaStatement, RyaDAOException> iter = queryEngine.query(new RyaStatement(cpu, loadPerc, null), queryConf); - int count = 0; - while (iter.hasNext()) { - iter.next().getObject(); - count++; - } - iter.close(); - assertEquals(limit, count); - } - - @Test - public void testAddValue() throws Exception { - RyaURI cpu = new RyaURI(litdupsNS + "cpu"); - RyaURI loadPerc = new RyaURI(litdupsNS + "loadPerc"); - RyaURI uri1 = new RyaURI(litdupsNS + "uri1"); - String myval = "myval"; - dao.add(new RyaStatement(cpu, loadPerc, uri1, null, null, null, myval.getBytes())); - - AccumuloRyaQueryEngine queryEngine = dao.getQueryEngine(); - CloseableIteration<RyaStatement, RyaDAOException> iter = queryEngine.query(new RyaStatement(cpu, loadPerc, null), conf); - assertTrue(iter.hasNext()); - assertEquals(myval, new String(iter.next().getValue())); - iter.close(); - } - - @Test - public void testAddCv() throws Exception { - RyaURI cpu = new RyaURI(litdupsNS + "cpu"); - RyaURI loadPerc = new RyaURI(litdupsNS + "loadPerc"); - RyaURI uri1 = new RyaURI(litdupsNS + "uri1"); - RyaURI uri2 = new RyaURI(litdupsNS + "uri2"); - RyaURI uri3 = new RyaURI(litdupsNS + "uri3"); - byte[] colVisABC = "A|B|C".getBytes(); - byte[] colVisAB = "A|B".getBytes(); - byte[] colVisA = "A".getBytes(); - dao.add(new RyaStatement(cpu, loadPerc, uri1, null, null, colVisABC)); - dao.add(new RyaStatement(cpu, loadPerc, uri2, null, null, colVisAB)); - dao.add(new RyaStatement(cpu, loadPerc, uri3, null, null, colVisA)); - - AccumuloRyaQueryEngine queryEngine = dao.getQueryEngine(); - - //query with no auth - CloseableIteration<RyaStatement, RyaDAOException> iter = queryEngine.query(new RyaStatement(cpu, loadPerc, null), conf); - int count = 0; - while (iter.hasNext()) { - count++; - iter.next(); - } - assertEquals(0, count); - iter.close(); - - AccumuloRdfConfiguration queryConf = new AccumuloRdfConfiguration(); - queryConf.setAuth("B"); - iter = queryEngine.query(new RyaStatement(cpu, loadPerc, null), queryConf); - count = 0; - while (iter.hasNext()) { - iter.next(); - count++; - } - iter.close(); - assertEquals(2, count); - - queryConf.setAuth("A"); - iter = queryEngine.query(new RyaStatement(cpu, loadPerc, null), queryConf); - count = 0; - while (iter.hasNext()) { - iter.next(); - count++; - } - iter.close(); - assertEquals(3, count); - } - - @Test - public void testTTL() throws Exception { - RyaURI cpu = new RyaURI(litdupsNS + "cpu"); - RyaURI loadPerc = new RyaURI(litdupsNS + "loadPerc"); - long current = System.currentTimeMillis(); - dao.add(new RyaStatement(cpu, loadPerc, new RyaURI(litdupsNS + "uri1"), null, null, null, null, current)); - dao.add(new RyaStatement(cpu, loadPerc, new RyaURI(litdupsNS + "uri2"), null, null, null, null, current - 1010l)); - dao.add(new RyaStatement(cpu, loadPerc, new RyaURI(litdupsNS + "uri3"), null, null, null, null, current - 2010l)); - dao.add(new RyaStatement(cpu, loadPerc, new RyaURI(litdupsNS + "uri4"), null, null, null, null, current - 3010l)); - dao.add(new RyaStatement(cpu, loadPerc, new RyaURI(litdupsNS + "uri5"), null, null, null, null, current - 4010l)); - - AccumuloRyaQueryEngine queryEngine = dao.getQueryEngine(); - AccumuloRdfConfiguration queryConf = conf.clone(); - queryConf.setTtl(3000l); - - CloseableIteration<RyaStatement, RyaDAOException> iter = queryEngine.query(new RyaStatement(cpu, loadPerc, null), queryConf); - int count = 0; - while (iter.hasNext()) { - iter.next().getObject(); - count++; - } - iter.close(); - assertEquals(3, count); - - queryConf.setStartTime(current - 3000l); - iter = queryEngine.query(new RyaStatement(cpu, loadPerc, null), queryConf); - count = 0; - while (iter.hasNext()) { - iter.next().getObject(); - count++; - } - iter.close(); - assertEquals(2, count); - } - - @Test - public void testGetNamespace() throws Exception { - dao.addNamespace("ns", litdupsNS); - assertEquals(litdupsNS, dao.getNamespace("ns")); - dao.removeNamespace("ns"); - assertNull(dao.getNamespace("ns")); - } - - //TOOD: Add test for set of queries - @Test - public void testQuery() throws Exception { - RyaURI cpu = new RyaURI(litdupsNS + "cpu"); - RyaURI loadPerc = new RyaURI(litdupsNS + "loadPerc"); - RyaURI uri1 = new RyaURI(litdupsNS + "uri1"); - RyaURI uri2 = new RyaURI(litdupsNS + "uri2"); - RyaURI uri3 = new RyaURI(litdupsNS + "uri3"); - RyaURI uri4 = new RyaURI(litdupsNS + "uri4"); - RyaURI uri5 = new RyaURI(litdupsNS + "uri5"); - RyaURI uri6 = new RyaURI(litdupsNS + "uri6"); - dao.add(new RyaStatement(cpu, loadPerc, uri1)); - dao.add(new RyaStatement(cpu, loadPerc, uri2)); - dao.add(new RyaStatement(cpu, loadPerc, uri3)); - dao.add(new RyaStatement(cpu, loadPerc, uri4)); - dao.add(new RyaStatement(cpu, loadPerc, uri5)); - dao.add(new RyaStatement(cpu, loadPerc, uri6)); - - AccumuloRyaQueryEngine queryEngine = dao.getQueryEngine(); - - Collection<RyaStatement> coll = new ArrayList(); - coll.add(new RyaStatement(null, loadPerc, uri1)); - coll.add(new RyaStatement(null, loadPerc, uri2)); - CloseableIteration<RyaStatement, RyaDAOException> iter = queryEngine.batchQuery(coll, conf); - int count = 0; - while (iter.hasNext()) { - count++; - iter.next(); - } - iter.close(); - assertEquals(2, count); - - //now use batchscanner - AccumuloRdfConfiguration queryConf = new AccumuloRdfConfiguration(conf); - queryConf.setMaxRangesForScanner(2); - - coll = new ArrayList(); - coll.add(new RyaStatement(null, loadPerc, uri1)); - coll.add(new RyaStatement(null, loadPerc, uri2)); - coll.add(new RyaStatement(null, loadPerc, uri3)); - coll.add(new RyaStatement(null, loadPerc, uri4)); - iter = queryEngine.batchQuery(coll, queryConf); - assertTrue(iter.hasNext()); //old code had a weird behaviour that could not perform hasNext consecutively - assertTrue(iter.hasNext()); - assertTrue(iter.hasNext()); - count = 0; - while (iter.hasNext()) { - count++; - assertTrue(iter.hasNext()); - iter.next(); - } - iter.close(); - assertEquals(4, count); - } - - @Test - public void testQueryDates() throws Exception { - RyaURI cpu = new RyaURI(litdupsNS + "cpu"); - RyaURI loadPerc = new RyaURI(litdupsNS + "loadPerc"); - RyaType uri0 = new RyaType(XMLSchema.DATETIME, "1960-01-01"); // How handles local time - RyaType uri1 = new RyaType(XMLSchema.DATETIME, "1992-01-01T+10:00"); // See Magadan Time - RyaType uri2 = new RyaType(XMLSchema.DATETIME, "2000-01-01TZ"); // How it handles UTC. - RyaType uri3 = new RyaType(XMLSchema.DATETIME, "2000-01-01T00:00:01.111Z"); - RyaType uri4 = new RyaType(XMLSchema.DATETIME, "2000-01-01T00:00:01.111Z"); // duplicate - RyaType uri5 = new RyaType(XMLSchema.DATETIME, "2000-01-01T00:00:01-00:00"); - RyaType uri6 = new RyaType(XMLSchema.DATETIME, "2000-01-01T00:00:01Z"); // duplicate - RyaType uri7 = new RyaType(XMLSchema.DATETIME, "-2000-01-01T00:00:01Z"); - RyaType uri8 = new RyaType(XMLSchema.DATETIME, "111-01-01T00:00:01Z"); - RyaType uri9 = new RyaType(XMLSchema.DATETIME, "12345-01-01T00:00:01Z"); - - dao.add(new RyaStatement(cpu, loadPerc, uri0)); - dao.add(new RyaStatement(cpu, loadPerc, uri1)); - dao.add(new RyaStatement(cpu, loadPerc, uri2)); - dao.add(new RyaStatement(cpu, loadPerc, uri3)); - dao.add(new RyaStatement(cpu, loadPerc, uri4)); - dao.add(new RyaStatement(cpu, loadPerc, uri5)); - dao.add(new RyaStatement(cpu, loadPerc, uri6)); - dao.add(new RyaStatement(cpu, loadPerc, uri7)); - dao.add(new RyaStatement(cpu, loadPerc, uri8)); - dao.add(new RyaStatement(cpu, loadPerc, uri9)); - - AccumuloRyaQueryEngine queryEngine = dao.getQueryEngine(); - - Collection<RyaStatement> coll = new ArrayList(); - coll.add(new RyaStatement(null, loadPerc, uri0)); - coll.add(new RyaStatement(null, loadPerc, uri1)); - coll.add(new RyaStatement(null, loadPerc, uri2)); - CloseableIteration<RyaStatement, RyaDAOException> iter = queryEngine.batchQuery(coll, conf); - int count = 0; - while (iter.hasNext()) { - count++; - iter.next(); - } - iter.close(); - assertEquals("Three time zones should be normalized when stored, then normalized same when queried.",3, count); - - //now use batchscanner - AccumuloRdfConfiguration queryConf = new AccumuloRdfConfiguration(conf); - queryConf.setMaxRangesForScanner(2); - - coll = new ArrayList(); - coll.add(new RyaStatement(null, loadPerc, uri0)); - coll.add(new RyaStatement(null, loadPerc, uri1)); - coll.add(new RyaStatement(null, loadPerc, uri2)); - coll.add(new RyaStatement(null, loadPerc, uri3)); - coll.add(new RyaStatement(null, loadPerc, uri4)); - coll.add(new RyaStatement(null, loadPerc, uri5)); - coll.add(new RyaStatement(null, loadPerc, uri6)); - coll.add(new RyaStatement(null, loadPerc, uri7)); - coll.add(new RyaStatement(null, loadPerc, uri8)); - coll.add(new RyaStatement(null, loadPerc, uri9)); - iter = queryEngine.batchQuery(coll, queryConf); - count = 0; - while (iter.hasNext()) { - count++; - iter.next(); - } - iter.close(); - assertEquals("Variety of time specs, including BC, pre-1970, duplicate pair ovewrite,future, 3 digit year.",8, count); - } - - @Test - public void testQueryCollectionRegex() throws Exception { - RyaURI cpu = new RyaURI(litdupsNS + "cpu"); - RyaURI loadPerc = new RyaURI(litdupsNS + "loadPerc"); - RyaURI uri1 = new RyaURI(litdupsNS + "uri1"); - RyaURI uri2 = new RyaURI(litdupsNS + "uri2"); - RyaURI uri3 = new RyaURI(litdupsNS + "uri3"); - RyaURI uri4 = new RyaURI(litdupsNS + "uri4"); - RyaURI uri5 = new RyaURI(litdupsNS + "uri5"); - RyaURI uri6 = new RyaURI(litdupsNS + "uri6"); - dao.add(new RyaStatement(cpu, loadPerc, uri1)); - dao.add(new RyaStatement(cpu, loadPerc, uri2)); - dao.add(new RyaStatement(cpu, loadPerc, uri3)); - dao.add(new RyaStatement(cpu, loadPerc, uri4)); - dao.add(new RyaStatement(cpu, loadPerc, uri5)); - dao.add(new RyaStatement(cpu, loadPerc, uri6)); - - AccumuloRyaQueryEngine queryEngine = dao.getQueryEngine(); - - Collection<RyaStatement> coll = new ArrayList(); - coll.add(new RyaStatement(null, loadPerc, uri1)); - coll.add(new RyaStatement(null, loadPerc, uri2)); - conf.setRegexPredicate(loadPerc.getData()); - CloseableIteration<RyaStatement, RyaDAOException> iter = queryEngine.batchQuery(coll, conf); - int count = 0; - while (iter.hasNext()) { - count++; - iter.next(); - } - iter.close(); - assertEquals(2, count); - - conf.setRegexPredicate("notLoadPerc"); - iter = queryEngine.batchQuery(coll, conf); - count = 0; - while (iter.hasNext()) { - count++; - iter.next(); - } - iter.close(); - assertEquals(0, count); - } - - @Test - public void testQueryCollectionRegexWBatchScanner() throws Exception { - RyaURI cpu = new RyaURI(litdupsNS + "cpu"); - RyaURI loadPerc = new RyaURI(litdupsNS + "loadPerc"); - RyaURI uri1 = new RyaURI(litdupsNS + "uri1"); - RyaURI uri2 = new RyaURI(litdupsNS + "uri2"); - RyaURI uri3 = new RyaURI(litdupsNS + "uri3"); - RyaURI uri4 = new RyaURI(litdupsNS + "uri4"); - RyaURI uri5 = new RyaURI(litdupsNS + "uri5"); - RyaURI uri6 = new RyaURI(litdupsNS + "uri6"); - dao.add(new RyaStatement(cpu, loadPerc, uri1)); - dao.add(new RyaStatement(cpu, loadPerc, uri2)); - dao.add(new RyaStatement(cpu, loadPerc, uri3)); - dao.add(new RyaStatement(cpu, loadPerc, uri4)); - dao.add(new RyaStatement(cpu, loadPerc, uri5)); - dao.add(new RyaStatement(cpu, loadPerc, uri6)); - - AccumuloRyaQueryEngine queryEngine = dao.getQueryEngine(); - AccumuloRdfConfiguration queryConf = new AccumuloRdfConfiguration(conf); - queryConf.setMaxRangesForScanner(1); - - Collection<RyaStatement> coll = new ArrayList(); - coll.add(new RyaStatement(null, loadPerc, uri1)); - coll.add(new RyaStatement(null, loadPerc, uri2)); - conf.setRegexPredicate(loadPerc.getData()); - CloseableIteration<RyaStatement, RyaDAOException> iter = queryEngine.batchQuery(coll, queryConf); - int count = 0; - while (iter.hasNext()) { - count++; - iter.next(); - } - iter.close(); - assertEquals(2, count); - - queryConf.setRegexPredicate("notLoadPerc"); - iter = queryEngine.batchQuery(coll, queryConf); - count = 0; - while (iter.hasNext()) { - count++; - iter.next(); - } - iter.close(); - assertEquals(0, count); - } - - @Test - public void testLiteralTypes() throws Exception { - RyaURI cpu = new RyaURI(litdupsNS + "cpu"); - RyaURI loadPerc = new RyaURI(litdupsNS + "loadPerc"); - RyaType longLit = new RyaType(XMLSchema.LONG, "3"); - - dao.add(new RyaStatement(cpu, loadPerc, longLit)); - - AccumuloRyaQueryEngine queryEngine = dao.getQueryEngine(); - - CloseableIteration<RyaStatement, RyaDAOException> query = queryEngine.query(new RyaStatement(cpu, null, null), conf); - assertTrue(query.hasNext()); - RyaStatement next = query.next(); - assertEquals(new Long(longLit.getData()), new Long(next.getObject().getData())); - query.close(); - - RyaType doubleLit = new RyaType(XMLSchema.DOUBLE, "2.0"); - - dao.add(new RyaStatement(cpu, loadPerc, doubleLit)); - - query = queryEngine.query(new RyaStatement(cpu, loadPerc, doubleLit), conf); - assertTrue(query.hasNext()); - next = query.next(); - assertEquals(Double.parseDouble(doubleLit.getData()), Double.parseDouble(next.getObject().getData()), 0.001); - query.close(); - } - - @Test - public void testSameLiteralStringTypes() throws Exception { - RyaURI cpu = new RyaURI(litdupsNS + "cpu"); - RyaURI loadPerc = new RyaURI(litdupsNS + "loadPerc"); - RyaType longLit = new RyaType(XMLSchema.LONG, "10"); - RyaType strLit = new RyaType(XMLSchema.STRING, new String(RyaContext.getInstance().serializeType(longLit)[0])); - - RyaStatement expected = new RyaStatement(cpu, loadPerc, longLit); - dao.add(expected); - dao.add(new RyaStatement(cpu, loadPerc, strLit)); - - AccumuloRyaQueryEngine queryEngine = dao.getQueryEngine(); - - CloseableIteration<RyaStatement, RyaDAOException> query = queryEngine.query(new RyaStatement(cpu, loadPerc, longLit), conf); - assertTrue(query.hasNext()); - RyaStatement next = query.next(); - assertEquals(new Long(longLit.getData()), new Long(next.getObject().getData())); - assertEquals(longLit.getDataType(), next.getObject().getDataType()); - assertFalse(query.hasNext()); - query.close(); - } - - @Test - public void testPurge() throws RyaDAOException, TableNotFoundException { - dao.add(newRyaStatement()); - assertFalse("table should not be empty", areTablesEmpty()); - - dao.purge(conf); - assertTrue("table should be empty", areTablesEmpty()); - //assertNotNull(dao.getVersion()); - } - - @Test - public void testPurgeDoesNotBreakBatchWriters() throws TableNotFoundException, RyaDAOException { - dao.purge(conf); - assertTrue("table should be empty", areTablesEmpty()); - - dao.add(newRyaStatement()); - assertFalse("table should not be empty", areTablesEmpty()); - } - - @Test - public void testDropAndDestroy() throws RyaDAOException { - assertTrue(dao.isInitialized()); - dao.dropAndDestroy(); - for (String tableName : dao.getTables()) { - assertFalse(tableExists(tableName)); - } - assertFalse(dao.isInitialized()); - } - - @Test - public void testQueryWithIterators() throws Exception { - RyaURI cpu = new RyaURI(litdupsNS + "cpu"); - RyaURI loadPerc = new RyaURI(litdupsNS + "loadPerc"); - RyaURI uri1 = new RyaURI(litdupsNS + "uri1"); - dao.add(new RyaStatement(cpu, loadPerc, uri1, null, "qual1")); - dao.add(new RyaStatement(cpu, loadPerc, uri1, null, "qual2")); - - AccumuloRyaQueryEngine queryEngine = dao.getQueryEngine(); - - AccumuloRdfConfiguration queryConf = new AccumuloRdfConfiguration(conf); - IteratorSetting firstEntryInRow = new IteratorSetting(3 /* correct value?? */, FirstEntryInRowIterator.class); - queryConf.setAdditionalIterators(firstEntryInRow); - - Collection<RyaStatement> coll = new ArrayList<>(); - coll.add(new RyaStatement(null, loadPerc, uri1)); - CloseableIteration<RyaStatement, RyaDAOException> iter = queryEngine.batchQuery(coll, queryConf); - int count = 0; - while (iter.hasNext()) { - count++; - iter.next(); - } - iter.close(); - assertEquals(1, count); - - //Assert that without the iterator we get 2 - coll = new ArrayList<>(); - coll.add(new RyaStatement(null, loadPerc, uri1)); - iter = queryEngine.batchQuery(coll, conf); - count = 0; - while (iter.hasNext()) { - count++; - iter.next(); - } - iter.close(); - assertEquals(2, count); - - } - - private boolean areTablesEmpty() throws TableNotFoundException { - for (String table : dao.getTables()) { - if (tableExists(table)) { - // TODO: filter out version - if (createScanner(table).iterator().hasNext()) { - return false; - } - } - } - return true; - } - - private boolean tableExists(String tableName) { - return dao.getConnector().tableOperations().exists(tableName); - } - - private Scanner createScanner(String tableName) throws TableNotFoundException { - return dao.getConnector().createScanner(tableName, conf.getAuthorizations()); - } - - private RyaStatement newRyaStatement() { - RyaURI subject = new RyaURI(litdupsNS + randomString()); - RyaURI predicate = new RyaURI(litdupsNS + randomString()); - RyaType object = new RyaType(randomString()); - - return new RyaStatement(subject, predicate, object); - } - - private String randomString() { - return UUID.randomUUID().toString(); - } -} http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/44a2dcf0/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/AccumuloRyaITBase.java ---------------------------------------------------------------------- diff --git a/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/AccumuloRyaITBase.java b/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/AccumuloRyaITBase.java deleted file mode 100644 index dea227b..0000000 --- a/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/AccumuloRyaITBase.java +++ /dev/null @@ -1,109 +0,0 @@ -/** - * Licensed to the Apache Software Foundation (ASF) under one - * or more contributor license agreements. See the NOTICE file - * distributed with this work for additional information - * regarding copyright ownership. The ASF licenses this file - * to you 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. - */ -package mvm.rya.accumulo; - -import java.io.IOException; -import java.util.Date; -import java.util.concurrent.atomic.AtomicInteger; - -import org.apache.accumulo.core.client.AccumuloException; -import org.apache.accumulo.core.client.AccumuloSecurityException; -import org.apache.accumulo.core.client.TableNotFoundException; -import org.junit.AfterClass; -import org.junit.Before; -import org.junit.BeforeClass; - -import com.google.common.base.Optional; - -import mvm.rya.accumulo.instance.AccumuloRyaInstanceDetailsRepository; -import mvm.rya.api.instance.RyaDetails; -import mvm.rya.api.instance.RyaDetails.EntityCentricIndexDetails; -import mvm.rya.api.instance.RyaDetails.FreeTextIndexDetails; -import mvm.rya.api.instance.RyaDetails.GeoIndexDetails; -import mvm.rya.api.instance.RyaDetails.JoinSelectivityDetails; -import mvm.rya.api.instance.RyaDetails.PCJIndexDetails; -import mvm.rya.api.instance.RyaDetails.ProspectorDetails; -import mvm.rya.api.instance.RyaDetails.TemporalIndexDetails; -import mvm.rya.api.instance.RyaDetailsRepository; -import mvm.rya.api.instance.RyaDetailsRepository.AlreadyInitializedException; -import mvm.rya.api.instance.RyaDetailsRepository.RyaDetailsRepositoryException; - -/** - * Contains boilerplate code for spinning up a Mini Accumulo Cluster and initializing - * some of the Rya stuff. We can not actually initialize an instance of Rya here - * because Sail is not available to us. - */ -public class AccumuloRyaITBase { - - // Managed the MiniAccumuloCluster - private static final MiniAccumuloClusterInstance cluster = new MiniAccumuloClusterInstance(); - - // Manage the Rya instances that are hosted on the cluster - protected static final AtomicInteger ryaInstanceNameCounter = new AtomicInteger(1); - private String ryaInstanceName; - - @BeforeClass - public static void initCluster() throws IOException, InterruptedException, AccumuloException, AccumuloSecurityException { - cluster.startMiniAccumulo(); - } - - @Before - public void prepareForNextTest() throws AccumuloException, AccumuloSecurityException, TableNotFoundException, AlreadyInitializedException, RyaDetailsRepositoryException { - // Get the next Rya instance name. - ryaInstanceName = "testInstance" + ryaInstanceNameCounter.getAndIncrement() + "_"; - - // Create Rya Details for the instance name. - final RyaDetailsRepository detailsRepo = new AccumuloRyaInstanceDetailsRepository(cluster.getConnector(), ryaInstanceName); - - final RyaDetails details = RyaDetails.builder() - .setRyaInstanceName(ryaInstanceName) - .setRyaVersion("0.0.0.0") - .setFreeTextDetails( new FreeTextIndexDetails(true) ) - .setEntityCentricIndexDetails( new EntityCentricIndexDetails(true) ) - .setGeoIndexDetails( new GeoIndexDetails(true) ) - .setTemporalIndexDetails( new TemporalIndexDetails(true) ) - .setPCJIndexDetails( - PCJIndexDetails.builder() - .setEnabled(true) ) - .setJoinSelectivityDetails( new JoinSelectivityDetails( Optional.<Date>absent() ) ) - .setProspectorDetails( new ProspectorDetails( Optional.<Date>absent() )) - .build(); - - detailsRepo.initialize(details); - } - - @AfterClass - public static void tearDownCluster() throws IOException, InterruptedException { - cluster.stopMiniAccumulo(); - } - - /** - * @return The {@link MiniAccumuloClusterInstance} used by the tests. - */ - public MiniAccumuloClusterInstance getClusterInstance() { - return cluster; - } - - /** - * @return The name of the Rya instance that is being used for the current test. - */ - public String getRyaInstanceName() { - return ryaInstanceName; - } -} \ No newline at end of file
