You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@rya.apache.org by pu...@apache.org on 2015/10/28 20:24:37 UTC
[43/51] [partial] incubator-rya git commit: now including the actual
project
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/KeyValueToRyaStatementFunction.java
----------------------------------------------------------------------
diff --git a/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/KeyValueToRyaStatementFunction.java b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/KeyValueToRyaStatementFunction.java
new file mode 100644
index 0000000..553624a
--- /dev/null
+++ b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/KeyValueToRyaStatementFunction.java
@@ -0,0 +1,71 @@
+package mvm.rya.accumulo.query;
+
+/*
+ * #%L
+ * mvm.rya.accumulo.rya
+ * %%
+ * Copyright (C) 2014 Rya
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import 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/92ddfa59/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/RangeBindingSetEntries.java
----------------------------------------------------------------------
diff --git a/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/RangeBindingSetEntries.java b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/RangeBindingSetEntries.java
new file mode 100644
index 0000000..894ed84
--- /dev/null
+++ b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/RangeBindingSetEntries.java
@@ -0,0 +1,57 @@
+package mvm.rya.accumulo.query;
+
+/*
+ * #%L
+ * mvm.rya.accumulo.rya
+ * %%
+ * Copyright (C) 2014 Rya
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import 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/92ddfa59/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/RyaStatementBindingSetKeyValueIterator.java
----------------------------------------------------------------------
diff --git a/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/RyaStatementBindingSetKeyValueIterator.java b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/RyaStatementBindingSetKeyValueIterator.java
new file mode 100644
index 0000000..8d90b0c
--- /dev/null
+++ b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/RyaStatementBindingSetKeyValueIterator.java
@@ -0,0 +1,153 @@
+package mvm.rya.accumulo.query;
+
+/*
+ * #%L
+ * mvm.rya.accumulo.rya
+ * %%
+ * Copyright (C) 2014 Rya
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import 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/92ddfa59/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/RyaStatementKeyValueIterator.java
----------------------------------------------------------------------
diff --git a/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/RyaStatementKeyValueIterator.java b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/RyaStatementKeyValueIterator.java
new file mode 100644
index 0000000..a4d0a40
--- /dev/null
+++ b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/RyaStatementKeyValueIterator.java
@@ -0,0 +1,106 @@
+package mvm.rya.accumulo.query;
+
+/*
+ * #%L
+ * mvm.rya.accumulo.rya
+ * %%
+ * Copyright (C) 2014 Rya
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import 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/92ddfa59/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/ScannerBaseCloseableIterable.java
----------------------------------------------------------------------
diff --git a/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/ScannerBaseCloseableIterable.java b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/ScannerBaseCloseableIterable.java
new file mode 100644
index 0000000..95d76b9
--- /dev/null
+++ b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/query/ScannerBaseCloseableIterable.java
@@ -0,0 +1,55 @@
+package mvm.rya.accumulo.query;
+
+/*
+ * #%L
+ * mvm.rya.accumulo.rya
+ * %%
+ * Copyright (C) 2014 Rya
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import 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/92ddfa59/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/utils/TimeRangeFilter.java
----------------------------------------------------------------------
diff --git a/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/utils/TimeRangeFilter.java b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/utils/TimeRangeFilter.java
new file mode 100644
index 0000000..a2381b2
--- /dev/null
+++ b/dao/accumulo.rya/src/main/java/mvm/rya/accumulo/utils/TimeRangeFilter.java
@@ -0,0 +1,86 @@
+package mvm.rya.accumulo.utils;
+
+/*
+ * #%L
+ * mvm.rya.accumulo.rya
+ * %%
+ * Copyright (C) 2014 Rya
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import 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;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/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
new file mode 100644
index 0000000..d1ddbaa
--- /dev/null
+++ b/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/AccumuloRdfConfigurationTest.java
@@ -0,0 +1,58 @@
+package mvm.rya.accumulo;
+
+/*
+ * #%L
+ * mvm.rya.accumulo.rya
+ * %%
+ * Copyright (C) 2014 Rya
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import org.apache.accumulo.core.security.Authorizations;
+import org.junit.Test;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Arrays;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Date: 1/28/13
+ * Time: 8:36 AM
+ */
+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());
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/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
new file mode 100644
index 0000000..9295dd9
--- /dev/null
+++ b/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/AccumuloRyaDAOTest.java
@@ -0,0 +1,664 @@
+package mvm.rya.accumulo;
+
+/*
+ * #%L
+ * mvm.rya.accumulo.rya
+ * %%
+ * Copyright (C) 2014 Rya
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import info.aduna.iteration.CloseableIteration;
+import mvm.rya.accumulo.query.AccumuloRyaQueryEngine;
+import mvm.rya.api.RdfCloudTripleStoreUtils;
+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.Scanner;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.mock.MockInstance;
+import org.calrissian.mango.collect.CloseableIterable;
+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;
+import org.openrdf.query.BindingSet;
+
+import java.util.*;
+
+import static org.junit.Assert.*;
+
+/**
+ * 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 uri1 = new RyaType(XMLSchema.DATETIME, "2000-01-01");
+ RyaType uri2 = new RyaType(XMLSchema.DATETIME, "2000-01-01TZ");
+ RyaType uri3 = new RyaType(XMLSchema.DATETIME, "2000-01-01T00:00:01.111Z");
+ RyaType uri4 = new RyaType(XMLSchema.DATETIME, "2000-01-01T00:00:01");
+ RyaType uri5 = new RyaType(XMLSchema.DATETIME, "2000-01-01T00:00:01.111");
+ RyaType uri6 = new RyaType(XMLSchema.DATETIME, "2000-01-01T00:00:01Z");
+ 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, 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, 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));
+ 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(9, 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());
+ }
+
+ 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/92ddfa59/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/DefineTripleQueryRangeFactoryTest.java
----------------------------------------------------------------------
diff --git a/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/DefineTripleQueryRangeFactoryTest.java b/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/DefineTripleQueryRangeFactoryTest.java
new file mode 100644
index 0000000..168e85c
--- /dev/null
+++ b/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/DefineTripleQueryRangeFactoryTest.java
@@ -0,0 +1,265 @@
+//package mvm.rya.accumulo;
+
+/*
+ * #%L
+ * mvm.rya.accumulo.rya
+ * %%
+ * Copyright (C) 2014 Rya
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+//
+//import junit.framework.TestCase;
+//import mvm.rya.accumulo.AccumuloRdfConfiguration;
+//import mvm.rya.accumulo.DefineTripleQueryRangeFactory;
+//import mvm.rya.accumulo.AccumuloRdfConfiguration;
+//import mvm.rya.accumulo.DefineTripleQueryRangeFactory;
+//import mvm.rya.api.domain.RangeValue;
+//import org.apache.accumulo.core.data.Range;
+//import org.openrdf.model.URI;
+//import org.openrdf.model.Value;
+//import org.openrdf.model.ValueFactory;
+//import org.openrdf.model.impl.ValueFactoryImpl;
+//
+//import java.util.Map;
+//
+//import static mvm.rya.api.RdfCloudTripleStoreConstants.*;
+//
+///**
+// */
+//public class DefineTripleQueryRangeFactoryTest extends TestCase {
+//
+// public static final String DELIM_BYTES_STR = new String(DELIM_BYTES);
+// public static final String URI_MARKER_STR = "\u0007";
+// public static final String RANGE_ENDKEY_SUFFIX = "\u0000";
+// DefineTripleQueryRangeFactory factory = new DefineTripleQueryRangeFactory();
+// ValueFactory vf = ValueFactoryImpl.getInstance();
+// static String litdupsNS = "urn:test:litdups#";
+//
+// private AccumuloRdfConfiguration conf = new AccumuloRdfConfiguration();
+//
+// public void testSPOCases() throws Exception {
+// URI cpu = vf.createURI(litdupsNS, "cpu");
+// URI loadPerc = vf.createURI(litdupsNS, "loadPerc");
+// URI obj = vf.createURI(litdupsNS, "uri1");
+//
+// //spo
+// Map.Entry<TABLE_LAYOUT, Range> entry =
+// factory.defineRange(cpu, loadPerc, obj, conf);
+// assertEquals(TABLE_LAYOUT.SPO, entry.getKey());
+// String expected_start = URI_MARKER_STR + cpu.stringValue() + DELIM_BYTES_STR +
+// URI_MARKER_STR + loadPerc.stringValue() + DELIM_BYTES_STR +
+// URI_MARKER_STR + obj.stringValue();
+// assertEquals(expected_start,
+// entry.getValue().getStartKey().getRow().toString());
+// assertEquals(expected_start + DELIM_STOP + RANGE_ENDKEY_SUFFIX,
+// entry.getValue().getEndKey().getRow().toString());
+//
+//
+// //sp
+// entry = factory.defineRange(cpu, loadPerc, null, conf);
+// assertEquals(TABLE_LAYOUT.SPO, entry.getKey());
+// expected_start = URI_MARKER_STR + cpu.stringValue() + DELIM_BYTES_STR +
+// URI_MARKER_STR + loadPerc.stringValue();
+// assertEquals(expected_start,
+// entry.getValue().getStartKey().getRow().toString());
+// assertEquals(expected_start + DELIM_STOP + RANGE_ENDKEY_SUFFIX,
+// entry.getValue().getEndKey().getRow().toString());
+//
+// //s
+// entry = factory.defineRange(cpu, null, null, conf);
+// assertEquals(TABLE_LAYOUT.SPO, entry.getKey());
+// expected_start = URI_MARKER_STR + cpu.stringValue();
+// assertEquals(expected_start,
+// entry.getValue().getStartKey().getRow().toString());
+// assertEquals(expected_start + DELIM_STOP + RANGE_ENDKEY_SUFFIX,
+// entry.getValue().getEndKey().getRow().toString());
+//
+// //all
+// entry = factory.defineRange(null, null, null, conf);
+// assertEquals(TABLE_LAYOUT.SPO, entry.getKey());
+// assertEquals("",
+// entry.getValue().getStartKey().getRow().toString());
+// assertEquals(new String(new byte[]{Byte.MAX_VALUE}) + DELIM_STOP + RANGE_ENDKEY_SUFFIX,
+// entry.getValue().getEndKey().getRow().toString());
+// }
+//
+// public void testSPOCasesWithRanges() throws Exception {
+// URI subj_start = vf.createURI(litdupsNS, "subj_start");
+// URI subj_end = vf.createURI(litdupsNS, "subj_stop");
+// URI pred_start = vf.createURI(litdupsNS, "pred_start");
+// URI pred_end = vf.createURI(litdupsNS, "pred_stop");
+// URI obj_start = vf.createURI(litdupsNS, "obj_start");
+// URI obj_end = vf.createURI(litdupsNS, "obj_stop");
+//
+// Value subj = new RangeValue(subj_start, subj_end);
+// Value pred = new RangeValue(pred_start, pred_end);
+// Value obj = new RangeValue(obj_start, obj_end);
+//
+// //spo - o has range
+// Map.Entry<TABLE_LAYOUT, Range> entry =
+// factory.defineRange(subj_start, pred_start, obj, conf);
+// assertEquals(TABLE_LAYOUT.SPO, entry.getKey());
+// String expected_start = URI_MARKER_STR + subj_start.stringValue() + DELIM_BYTES_STR +
+// URI_MARKER_STR + pred_start.stringValue() + DELIM_BYTES_STR +
+// URI_MARKER_STR + obj_start.stringValue();
+// assertEquals(expected_start,
+// entry.getValue().getStartKey().getRow().toString());
+// String expected_end = URI_MARKER_STR + subj_start.stringValue() + DELIM_BYTES_STR +
+// URI_MARKER_STR + pred_start.stringValue() + DELIM_BYTES_STR +
+// URI_MARKER_STR + obj_end.stringValue();
+// assertEquals(expected_end + DELIM_STOP + RANGE_ENDKEY_SUFFIX,
+// entry.getValue().getEndKey().getRow().toString());
+//
+// //sp - p has range
+// entry = factory.defineRange(subj_start, pred, null, conf);
+// assertEquals(TABLE_LAYOUT.SPO, entry.getKey());
+// expected_start = URI_MARKER_STR + subj_start.stringValue() + DELIM_BYTES_STR +
+// URI_MARKER_STR + pred_start.stringValue();
+// assertEquals(expected_start,
+// entry.getValue().getStartKey().getRow().toString());
+// expected_end = URI_MARKER_STR + subj_start.stringValue() + DELIM_BYTES_STR +
+// URI_MARKER_STR + pred_end.stringValue();
+// assertEquals(expected_end + DELIM_STOP + RANGE_ENDKEY_SUFFIX,
+// entry.getValue().getEndKey().getRow().toString());
+//
+// //s - s has range
+// entry = factory.defineRange(subj, null, null, conf);
+// assertEquals(TABLE_LAYOUT.SPO, entry.getKey());
+// expected_start = URI_MARKER_STR + subj_start.stringValue();
+// assertEquals(expected_start,
+// entry.getValue().getStartKey().getRow().toString());
+// expected_end = URI_MARKER_STR + subj_end.stringValue();
+// assertEquals(expected_end + DELIM_STOP + RANGE_ENDKEY_SUFFIX,
+// entry.getValue().getEndKey().getRow().toString());
+// }
+//
+// public void testPOCases() throws Exception {
+// URI loadPerc = vf.createURI(litdupsNS, "loadPerc");
+// URI obj = vf.createURI(litdupsNS, "uri1");
+//
+// //po
+// Map.Entry<TABLE_LAYOUT, Range> entry =
+// factory.defineRange(null, loadPerc, obj, conf);
+// assertEquals(TABLE_LAYOUT.PO, entry.getKey());
+// String expected_start = URI_MARKER_STR + loadPerc.stringValue() + DELIM_BYTES_STR +
+// URI_MARKER_STR + obj.stringValue();
+// assertEquals(expected_start,
+// entry.getValue().getStartKey().getRow().toString());
+// assertEquals(expected_start + DELIM_STOP + RANGE_ENDKEY_SUFFIX,
+// entry.getValue().getEndKey().getRow().toString());
+//
+// //p
+// entry = factory.defineRange(null, loadPerc, null, conf);
+// assertEquals(TABLE_LAYOUT.PO, entry.getKey());
+// expected_start = URI_MARKER_STR + loadPerc.stringValue();
+// assertEquals(expected_start,
+// entry.getValue().getStartKey().getRow().toString());
+// assertEquals(expected_start + DELIM_STOP + RANGE_ENDKEY_SUFFIX,
+// entry.getValue().getEndKey().getRow().toString());
+// }
+//
+// public void testPOCasesWithRanges() throws Exception {
+// URI pred_start = vf.createURI(litdupsNS, "pred_start");
+// URI pred_end = vf.createURI(litdupsNS, "pred_stop");
+// URI obj_start = vf.createURI(litdupsNS, "obj_start");
+// URI obj_end = vf.createURI(litdupsNS, "obj_stop");
+//
+// Value pred = new RangeValue(pred_start, pred_end);
+// Value obj = new RangeValue(obj_start, obj_end);
+//
+// //po
+// Map.Entry<TABLE_LAYOUT, Range> entry =
+// factory.defineRange(null, pred_start, obj, conf);
+// assertEquals(TABLE_LAYOUT.PO, entry.getKey());
+// String expected_start = URI_MARKER_STR + pred_start.stringValue() + DELIM_BYTES_STR +
+// URI_MARKER_STR + obj_start.stringValue();
+// assertEquals(expected_start,
+// entry.getValue().getStartKey().getRow().toString());
+// String expected_end = URI_MARKER_STR + pred_start.stringValue() + DELIM_BYTES_STR +
+// URI_MARKER_STR + obj_end.stringValue();
+// assertEquals(expected_end + DELIM_STOP + RANGE_ENDKEY_SUFFIX,
+// entry.getValue().getEndKey().getRow().toString());
+//
+// //p
+// entry = factory.defineRange(null, pred, null, conf);
+// assertEquals(TABLE_LAYOUT.PO, entry.getKey());
+// expected_start = URI_MARKER_STR + pred_start.stringValue();
+// assertEquals(expected_start,
+// entry.getValue().getStartKey().getRow().toString());
+// expected_end = URI_MARKER_STR + pred_end.stringValue();
+// assertEquals(expected_end + DELIM_STOP + RANGE_ENDKEY_SUFFIX,
+// entry.getValue().getEndKey().getRow().toString());
+// }
+//
+// public void testOSPCases() throws Exception {
+// URI cpu = vf.createURI(litdupsNS, "cpu");
+// URI obj = vf.createURI(litdupsNS, "uri1");
+//
+// //so
+// Map.Entry<TABLE_LAYOUT, Range> entry =
+// factory.defineRange(cpu, null, obj, conf);
+// assertEquals(TABLE_LAYOUT.OSP, entry.getKey());
+// String expected_start = URI_MARKER_STR + obj.stringValue() + DELIM_BYTES_STR +
+// URI_MARKER_STR + cpu.stringValue();
+// assertEquals(expected_start,
+// entry.getValue().getStartKey().getRow().toString());
+// assertEquals(expected_start + DELIM_STOP + RANGE_ENDKEY_SUFFIX,
+// entry.getValue().getEndKey().getRow().toString());
+//
+// //o
+// entry = factory.defineRange(null, null, obj, conf);
+// assertEquals(TABLE_LAYOUT.OSP, entry.getKey());
+// expected_start = URI_MARKER_STR + obj.stringValue();
+// assertEquals(expected_start,
+// entry.getValue().getStartKey().getRow().toString());
+// assertEquals(expected_start + DELIM_STOP + RANGE_ENDKEY_SUFFIX,
+// entry.getValue().getEndKey().getRow().toString());
+// }
+//
+//
+// public void testOSPCasesWithRanges() throws Exception {
+// URI subj_start = vf.createURI(litdupsNS, "subj_start");
+// URI subj_end = vf.createURI(litdupsNS, "subj_stop");
+// URI obj_start = vf.createURI(litdupsNS, "obj_start");
+// URI obj_end = vf.createURI(litdupsNS, "obj_stop");
+//
+// Value subj = new RangeValue(subj_start, subj_end);
+// Value obj = new RangeValue(obj_start, obj_end);
+//
+// //so - s should be the range
+// Map.Entry<TABLE_LAYOUT, Range> entry =
+// factory.defineRange(subj, null, obj_start, conf);
+// assertEquals(TABLE_LAYOUT.OSP, entry.getKey());
+// String expected_start = URI_MARKER_STR + obj_start.stringValue() + DELIM_BYTES_STR +
+// URI_MARKER_STR + subj_start.stringValue();
+// assertEquals(expected_start,
+// entry.getValue().getStartKey().getRow().toString());
+// String expected_end = URI_MARKER_STR + obj_start.stringValue() + DELIM_BYTES_STR +
+// URI_MARKER_STR + subj_end.stringValue();
+// assertEquals(expected_end + DELIM_STOP + RANGE_ENDKEY_SUFFIX,
+// entry.getValue().getEndKey().getRow().toString());
+//
+// //o - o is range
+// entry = factory.defineRange(null, null, obj, conf);
+// assertEquals(TABLE_LAYOUT.OSP, entry.getKey());
+// expected_start = URI_MARKER_STR + obj_start.stringValue();
+// assertEquals(expected_start,
+// entry.getValue().getStartKey().getRow().toString());
+// expected_end = URI_MARKER_STR + obj_end.stringValue();
+// assertEquals(expected_end + DELIM_STOP + RANGE_ENDKEY_SUFFIX,
+// entry.getValue().getEndKey().getRow().toString());
+// }
+//
+//}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/mr/eval/AccumuloRdfCountToolTest.java
----------------------------------------------------------------------
diff --git a/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/mr/eval/AccumuloRdfCountToolTest.java b/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/mr/eval/AccumuloRdfCountToolTest.java
new file mode 100644
index 0000000..07e7287
--- /dev/null
+++ b/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/mr/eval/AccumuloRdfCountToolTest.java
@@ -0,0 +1,281 @@
+package mvm.rya.accumulo.mr.eval;
+
+/*
+ * #%L
+ * mvm.rya.accumulo.rya
+ * %%
+ * Copyright (C) 2014 Rya
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import mvm.rya.accumulo.AccumuloRdfConfiguration;
+import mvm.rya.accumulo.AccumuloRyaDAO;
+import mvm.rya.api.RdfCloudTripleStoreConstants;
+import mvm.rya.api.domain.RyaStatement;
+import mvm.rya.api.domain.RyaURI;
+import mvm.rya.api.resolver.RdfToRyaConversions;
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.admin.SecurityOperations;
+import org.apache.accumulo.core.client.mock.MockInstance;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.PartialKey;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.security.TablePermission;
+import org.apache.hadoop.io.Text;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Ignore;
+import org.junit.Test;
+import org.openrdf.model.ValueFactory;
+import org.openrdf.model.impl.ValueFactoryImpl;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
+
+/**
+ * Created by IntelliJ IDEA.
+ * Date: 4/24/12
+ * Time: 5:05 PM
+ * To change this template use File | Settings | File Templates.
+ */
+@Ignore
+public class AccumuloRdfCountToolTest {
+
+ private String user = "user";
+ private String pwd = "pwd";
+ private String instance = "myinstance";
+ private String tablePrefix = "t_";
+ private Authorizations auths = Constants.NO_AUTHS;
+ private Connector connector;
+
+ private AccumuloRyaDAO dao;
+ private ValueFactory vf = new ValueFactoryImpl();
+ private AccumuloRdfConfiguration conf = new AccumuloRdfConfiguration();
+ static String litdupsNS = "urn:test:litdups#";
+
+ @Before
+ public void setUp() throws Exception {
+ connector = new MockInstance(instance).getConnector(user, pwd.getBytes());
+ connector.tableOperations().create(tablePrefix + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX);
+ connector.tableOperations().create(tablePrefix + RdfCloudTripleStoreConstants.TBL_PO_SUFFIX);
+ connector.tableOperations().create(tablePrefix + RdfCloudTripleStoreConstants.TBL_OSP_SUFFIX);
+ connector.tableOperations().create(tablePrefix + RdfCloudTripleStoreConstants.TBL_NS_SUFFIX);
+ connector.tableOperations().create(tablePrefix + RdfCloudTripleStoreConstants.TBL_EVAL_SUFFIX);
+ SecurityOperations secOps = connector.securityOperations();
+ secOps.createUser(user, pwd.getBytes(), auths);
+ secOps.grantTablePermission(user, tablePrefix + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, TablePermission.READ);
+ secOps.grantTablePermission(user, tablePrefix + RdfCloudTripleStoreConstants.TBL_PO_SUFFIX, TablePermission.READ);
+ secOps.grantTablePermission(user, tablePrefix + RdfCloudTripleStoreConstants.TBL_OSP_SUFFIX, TablePermission.READ);
+ secOps.grantTablePermission(user, tablePrefix + RdfCloudTripleStoreConstants.TBL_NS_SUFFIX, TablePermission.READ);
+ secOps.grantTablePermission(user, tablePrefix + RdfCloudTripleStoreConstants.TBL_EVAL_SUFFIX, TablePermission.READ);
+ secOps.grantTablePermission(user, tablePrefix + RdfCloudTripleStoreConstants.TBL_EVAL_SUFFIX, TablePermission.WRITE);
+
+ dao = new AccumuloRyaDAO();
+ dao.setConnector(connector);
+ conf.setTablePrefix(tablePrefix);
+ dao.setConf(conf);
+ dao.init();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ dao.destroy();
+ connector.tableOperations().delete(tablePrefix + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX);
+ connector.tableOperations().delete(tablePrefix + RdfCloudTripleStoreConstants.TBL_PO_SUFFIX);
+ connector.tableOperations().delete(tablePrefix + RdfCloudTripleStoreConstants.TBL_OSP_SUFFIX);
+ connector.tableOperations().delete(tablePrefix + RdfCloudTripleStoreConstants.TBL_NS_SUFFIX);
+ connector.tableOperations().delete(tablePrefix + RdfCloudTripleStoreConstants.TBL_EVAL_SUFFIX);
+ }
+
+ @Test
+ public void testMR() throws Exception {
+ RyaURI test1 = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "test1"));
+ RyaURI pred1 = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "pred1"));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(0))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(1))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(2))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(3))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(4))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(5))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(6))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(7))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(8))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(9))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(10))));
+
+ AccumuloRdfCountTool.main(new String[]{
+ "-Dac.mock=true",
+ "-Dac.instance=" + instance,
+ "-Dac.username=" + user,
+ "-Dac.pwd=" + pwd,
+ "-Drdf.tablePrefix=" + tablePrefix,
+ });
+
+ Map<String, Key> expectedValues = new HashMap<String, Key>();
+ String row = test1.getData();
+ expectedValues.put(row,
+ new Key(new Text(row),
+ RdfCloudTripleStoreConstants.SUBJECT_CF_TXT,
+ RdfCloudTripleStoreConstants.EMPTY_TEXT));
+ row = pred1.getData();
+ expectedValues.put(row,
+ new Key(new Text(row),
+ RdfCloudTripleStoreConstants.PRED_CF_TXT,
+ RdfCloudTripleStoreConstants.EMPTY_TEXT));
+ Scanner scanner = connector.createScanner(tablePrefix + RdfCloudTripleStoreConstants.TBL_EVAL_SUFFIX, auths);
+ scanner.setRange(new Range());
+ int count = 0;
+ for (Map.Entry<Key, Value> entry : scanner) {
+ assertTrue(expectedValues.get(entry.getKey().getRow().toString()).equals(entry.getKey(), PartialKey.ROW_COLFAM_COLQUAL));
+ assertEquals(11, Long.parseLong(entry.getValue().toString()));
+ count++;
+ }
+ assertEquals(2, count);
+ }
+
+// public void testMRObject() throws Exception {
+// URI pred1 = vf.createURI(litdupsNS, "pred1");
+// Literal literal = vf.createLiteral(0);
+// dao.add(new StatementImpl(vf.createURI(litdupsNS, "test0"), pred1, literal));
+// dao.add(new StatementImpl(vf.createURI(litdupsNS, "test1"), pred1, literal));
+// dao.add(new StatementImpl(vf.createURI(litdupsNS, "test2"), pred1, literal));
+// dao.add(new StatementImpl(vf.createURI(litdupsNS, "test3"), pred1, literal));
+// dao.add(new StatementImpl(vf.createURI(litdupsNS, "test4"), pred1, literal));
+// dao.add(new StatementImpl(vf.createURI(litdupsNS, "test5"), pred1, literal));
+// dao.add(new StatementImpl(vf.createURI(litdupsNS, "test6"), pred1, literal));
+// dao.add(new StatementImpl(vf.createURI(litdupsNS, "test7"), pred1, literal));
+// dao.add(new StatementImpl(vf.createURI(litdupsNS, "test8"), pred1, literal));
+// dao.add(new StatementImpl(vf.createURI(litdupsNS, "test9"), pred1, literal));
+// dao.add(new StatementImpl(vf.createURI(litdupsNS, "test10"), pred1, literal));
+// dao.commit();
+//
+// AccumuloRdfCountTool.main(new String[]{
+// "-Dac.mock=true",
+// "-Dac.instance=" + instance,
+// "-Dac.username=" + user,
+// "-Dac.pwd=" + pwd,
+// "-Drdf.tablePrefix=" + tablePrefix,
+// });
+//
+// Map<String, Key> expectedValues = new HashMap<String, Key>();
+// byte[] row_bytes = RdfCloudTripleStoreUtils.writeValue(literal);
+// expectedValues.put(new String(row_bytes),
+// new Key(new Text(row_bytes),
+// RdfCloudTripleStoreConstants.OBJ_CF_TXT,
+// RdfCloudTripleStoreConstants.INFO_TXT));
+// row_bytes = RdfCloudTripleStoreUtils.writeValue(pred1);
+// expectedValues.put(new String(row_bytes),
+// new Key(new Text(row_bytes),
+// RdfCloudTripleStoreConstants.PRED_CF_TXT,
+// RdfCloudTripleStoreConstants.INFO_TXT));
+// Scanner scanner = connector.createScanner(tablePrefix + RdfCloudTripleStoreConstants.TBL_EVAL_SUFFIX, auths);
+// scanner.setRange(new Range());
+// int count = 0;
+// for (Map.Entry<Key, Value> entry : scanner) {
+// assertTrue(expectedValues.get(entry.getKey().getRow().toString()).equals(entry.getKey(), PartialKey.ROW_COLFAM_COLQUAL));
+// assertEquals(11, Long.parseLong(entry.getValue().toString()));
+// count++;
+// }
+// assertEquals(2, count);
+// }
+
+ @Test
+ public void testTTL() throws Exception {
+ RyaURI test1 = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "test1"));
+ RyaURI pred1 = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "pred1"));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(0))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(1))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(2))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(3))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(4))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(5))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(6))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(7))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(8))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(9))));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(10))));
+
+ AccumuloRdfCountTool.main(new String[]{
+ "-Dac.mock=true",
+ "-Dac.instance=" + instance,
+ "-Dac.username=" + user,
+ "-Dac.pwd=" + pwd,
+ "-Dac.ttl=0",
+ "-Drdf.tablePrefix=" + tablePrefix,
+ });
+
+ Scanner scanner = connector.createScanner(tablePrefix + RdfCloudTripleStoreConstants.TBL_EVAL_SUFFIX, auths);
+ scanner.setRange(new Range());
+ int count = 0;
+ for (Map.Entry<Key, Value> entry : scanner) {
+ count++;
+ }
+ assertEquals(0, count);
+ }
+
+ @Test
+ public void testContext() throws Exception {
+ RyaURI test1 = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "test1"));
+ RyaURI pred1 = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "pred1"));
+ RyaURI cntxt = RdfToRyaConversions.convertURI(vf.createURI(litdupsNS, "cntxt"));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(0)), cntxt));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(1)), cntxt));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(2)), cntxt));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(3)), cntxt));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(4)), cntxt));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(5)), cntxt));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(6)), cntxt));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(7)), cntxt));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(8)), cntxt));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(9)), cntxt));
+ dao.add(new RyaStatement(test1, pred1, RdfToRyaConversions.convertLiteral(vf.createLiteral(10)), cntxt));
+
+ AccumuloRdfCountTool.main(new String[]{
+ "-Dac.mock=true",
+ "-Dac.instance=" + instance,
+ "-Dac.username=" + user,
+ "-Dac.pwd=" + pwd,
+ "-Drdf.tablePrefix=" + tablePrefix,
+ });
+
+ Map<String, Key> expectedValues = new HashMap<String, Key>();
+ String row = test1.getData();
+ expectedValues.put(row,
+ new Key(new Text(row),
+ RdfCloudTripleStoreConstants.SUBJECT_CF_TXT,
+ new Text(cntxt.getData())));
+ row = pred1.getData();
+ expectedValues.put(row,
+ new Key(new Text(row),
+ RdfCloudTripleStoreConstants.PRED_CF_TXT,
+ new Text(cntxt.getData())));
+ Scanner scanner = connector.createScanner(tablePrefix + RdfCloudTripleStoreConstants.TBL_EVAL_SUFFIX, auths);
+ scanner.setRange(new Range());
+ int count = 0;
+ for (Map.Entry<Key, Value> entry : scanner) {
+ assertTrue(expectedValues.get(entry.getKey().getRow().toString()).equals(entry.getKey(), PartialKey.ROW_COLFAM_COLQUAL));
+ assertEquals(11, Long.parseLong(entry.getValue().toString()));
+ count++;
+ }
+ assertEquals(2, count);
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/mr/fileinput/RdfFileInputToolTest.java
----------------------------------------------------------------------
diff --git a/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/mr/fileinput/RdfFileInputToolTest.java b/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/mr/fileinput/RdfFileInputToolTest.java
new file mode 100644
index 0000000..0d24b62
--- /dev/null
+++ b/dao/accumulo.rya/src/test/java/mvm/rya/accumulo/mr/fileinput/RdfFileInputToolTest.java
@@ -0,0 +1,145 @@
+package mvm.rya.accumulo.mr.fileinput;
+
+/*
+ * #%L
+ * mvm.rya.accumulo.rya
+ * %%
+ * Copyright (C) 2014 Rya
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import java.util.Iterator;
+import java.util.Map;
+
+import junit.framework.TestCase;
+import mvm.rya.accumulo.AccumuloRdfConfiguration;
+import mvm.rya.api.RdfCloudTripleStoreConstants;
+import mvm.rya.api.RdfCloudTripleStoreConstants.TABLE_LAYOUT;
+import mvm.rya.api.domain.RyaStatement;
+import mvm.rya.api.domain.RyaType;
+import mvm.rya.api.domain.RyaURI;
+import mvm.rya.api.resolver.RyaContext;
+import mvm.rya.api.resolver.RyaTripleContext;
+import mvm.rya.api.resolver.triple.TripleRow;
+
+import org.apache.accumulo.core.Constants;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.admin.SecurityOperations;
+import org.apache.accumulo.core.client.mock.MockInstance;
+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.security.Authorizations;
+import org.apache.accumulo.core.security.TablePermission;
+import org.apache.hadoop.io.Text;
+import org.openrdf.model.ValueFactory;
+import org.openrdf.model.impl.ValueFactoryImpl;
+import org.openrdf.rio.RDFFormat;
+
+/**
+ * Created by IntelliJ IDEA.
+ * Date: 4/25/12
+ * Time: 10:51 AM
+ * To change this template use File | Settings | File Templates.
+ */
+public class RdfFileInputToolTest extends TestCase {
+
+ private String user = "user";
+ private String pwd = "pwd";
+ private String instance = "myinstance";
+ private String tablePrefix = "t_";
+ private Authorizations auths = Constants.NO_AUTHS;
+ private Connector connector;
+
+ @Override
+ public void setUp() throws Exception {
+ super.setUp();
+ connector = new MockInstance(instance).getConnector(user, pwd.getBytes());
+ connector.tableOperations().create(tablePrefix + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX);
+ connector.tableOperations().create(tablePrefix + RdfCloudTripleStoreConstants.TBL_PO_SUFFIX);
+ connector.tableOperations().create(tablePrefix + RdfCloudTripleStoreConstants.TBL_OSP_SUFFIX);
+ connector.tableOperations().create(tablePrefix + RdfCloudTripleStoreConstants.TBL_NS_SUFFIX);
+ connector.tableOperations().create(tablePrefix + RdfCloudTripleStoreConstants.TBL_EVAL_SUFFIX);
+ SecurityOperations secOps = connector.securityOperations();
+ secOps.createUser(user, pwd.getBytes(), auths);
+ secOps.grantTablePermission(user, tablePrefix + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, TablePermission.READ);
+ secOps.grantTablePermission(user, tablePrefix + RdfCloudTripleStoreConstants.TBL_PO_SUFFIX, TablePermission.READ);
+ secOps.grantTablePermission(user, tablePrefix + RdfCloudTripleStoreConstants.TBL_OSP_SUFFIX, TablePermission.READ);
+ secOps.grantTablePermission(user, tablePrefix + RdfCloudTripleStoreConstants.TBL_NS_SUFFIX, TablePermission.READ);
+ secOps.grantTablePermission(user, tablePrefix + RdfCloudTripleStoreConstants.TBL_EVAL_SUFFIX, TablePermission.READ);
+ secOps.grantTablePermission(user, tablePrefix + RdfCloudTripleStoreConstants.TBL_EVAL_SUFFIX, TablePermission.WRITE);
+ }
+
+ @Override
+ public void tearDown() throws Exception {
+ super.tearDown();
+ connector.tableOperations().delete(tablePrefix + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX);
+ connector.tableOperations().delete(tablePrefix + RdfCloudTripleStoreConstants.TBL_PO_SUFFIX);
+ connector.tableOperations().delete(tablePrefix + RdfCloudTripleStoreConstants.TBL_OSP_SUFFIX);
+ connector.tableOperations().delete(tablePrefix + RdfCloudTripleStoreConstants.TBL_NS_SUFFIX);
+ connector.tableOperations().delete(tablePrefix + RdfCloudTripleStoreConstants.TBL_EVAL_SUFFIX);
+ }
+
+ public void testNTriplesInput() throws Exception {
+ RdfFileInputTool.main(new String[]{
+ "-Dac.mock=true",
+ "-Dac.instance=" + instance,
+ "-Dac.username=" + user,
+ "-Dac.pwd=" + pwd,
+ "-Drdf.tablePrefix=" + tablePrefix,
+ "-Drdf.format=" + RDFFormat.NTRIPLES.getName(),
+ "src/test/resources/test.ntriples",
+ });
+
+ Scanner scanner = connector.createScanner(tablePrefix + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, auths);
+ scanner.setRange(new Range());
+ Iterator<Map.Entry<Key, Value>> iterator = scanner.iterator();
+ ValueFactory vf = new ValueFactoryImpl();
+ assertTrue(iterator.hasNext());
+ RyaStatement rs = new RyaStatement(new RyaURI("urn:lubm:rdfts#GraduateStudent01"),
+ new RyaURI("urn:lubm:rdfts#hasFriend"),
+ new RyaURI("urn:lubm:rdfts#GraduateStudent02"));
+ assertEquals(new Text(RyaTripleContext.getInstance(new AccumuloRdfConfiguration()).serializeTriple(rs).get(TABLE_LAYOUT.SPO).getRow()), iterator.next().getKey().getRow());
+ }
+
+ public void testInputContext() throws Exception {
+ RdfFileInputTool.main(new String[]{
+ "-Dac.mock=true",
+ "-Dac.instance=" + instance,
+ "-Dac.username=" + user,
+ "-Dac.pwd=" + pwd,
+ "-Drdf.tablePrefix=" + tablePrefix,
+ "-Drdf.format=" + RDFFormat.TRIG.getName(),
+ "src/test/resources/namedgraphs.trig",
+ });
+
+ Scanner scanner = connector.createScanner(tablePrefix + RdfCloudTripleStoreConstants.TBL_SPO_SUFFIX, auths);
+ scanner.setRange(new Range());
+ Iterator<Map.Entry<Key, Value>> iterator = scanner.iterator();
+ ValueFactory vf = new ValueFactoryImpl();
+ assertTrue(iterator.hasNext());
+ RyaStatement rs = new RyaStatement(new RyaURI("http://www.example.org/exampleDocument#Monica"),
+ new RyaURI("http://www.example.org/vocabulary#name"),
+ new RyaType("Monica Murphy"),
+ new RyaURI("http://www.example.org/exampleDocument#G1"));
+ Key key = iterator.next().getKey();
+
+ TripleRow tripleRow = RyaTripleContext.getInstance(new AccumuloRdfConfiguration()).serializeTriple(rs).get(TABLE_LAYOUT.SPO);
+ assertEquals(new Text(tripleRow.getRow()), key.getRow());
+ assertEquals(new Text(tripleRow.getColumnFamily()), key.getColumnFamily());
+ }
+
+}