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:07 UTC
[13/51] [partial] incubator-rya git commit: now including the actual
project
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/extras/rya.prospector/src/main/java/mvm/rya/joinselect/mr/utils/JoinSelectStatsUtil.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/java/mvm/rya/joinselect/mr/utils/JoinSelectStatsUtil.java b/extras/rya.prospector/src/main/java/mvm/rya/joinselect/mr/utils/JoinSelectStatsUtil.java
new file mode 100644
index 0000000..59e7611
--- /dev/null
+++ b/extras/rya.prospector/src/main/java/mvm/rya/joinselect/mr/utils/JoinSelectStatsUtil.java
@@ -0,0 +1,182 @@
+package mvm.rya.joinselect.mr.utils;
+
+/*
+ * #%L
+ * mvm.rya.rya.prospector
+ * %%
+ * Copyright (C) 2014 - 2015 Rya
+ * %%
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ * #L%
+ */
+
+import static mvm.rya.accumulo.AccumuloRdfConstants.EMPTY_CV;
+import static mvm.rya.accumulo.AccumuloRdfConstants.EMPTY_VALUE;
+import static mvm.rya.api.RdfCloudTripleStoreConstants.EMPTY_TEXT;
+import static mvm.rya.joinselect.mr.utils.JoinSelectConstants.INSTANCE;
+import static mvm.rya.joinselect.mr.utils.JoinSelectConstants.PASSWORD;
+import static mvm.rya.joinselect.mr.utils.JoinSelectConstants.USERNAME;
+import static mvm.rya.joinselect.mr.utils.JoinSelectConstants.ZOOKEEPERS;
+
+import java.io.IOException;
+
+import mvm.rya.api.resolver.triple.TripleRow;
+
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.mapreduce.AccumuloInputFormat;
+import org.apache.accumulo.core.client.mapreduce.AccumuloOutputFormat;
+import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.accumulo.core.security.ColumnVisibility;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.lib.input.MultipleInputs;
+import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
+
+public class JoinSelectStatsUtil {
+
+ public static void initSumMRJob(Job job, String inputPath, String outtable, String auths) throws AccumuloSecurityException, IOException {
+ Configuration conf = job.getConfiguration();
+ String username = conf.get(USERNAME);
+ String password = conf.get(PASSWORD);
+ String instance = conf.get(INSTANCE);
+ String zookeepers = conf.get(ZOOKEEPERS);
+
+ if (zookeepers != null) {
+ AccumuloOutputFormat.setConnectorInfo(job, username, new PasswordToken(password));
+ AccumuloOutputFormat.setZooKeeperInstance(job, instance, zookeepers);
+ } else {
+ throw new IllegalArgumentException("Must specify zookeepers");
+ }
+
+ SequenceFileInputFormat.addInputPath(job, new Path(inputPath));
+ job.setInputFormatClass(SequenceFileInputFormat.class);
+ job.setMapOutputKeyClass(TripleEntry.class);
+ job.setMapOutputValueClass(CardList.class);
+
+ AccumuloOutputFormat.setDefaultTableName(job, outtable);
+ job.setOutputFormatClass(AccumuloOutputFormat.class);
+ job.setOutputKeyClass(Text.class);
+ job.setOutputValueClass(Mutation.class);
+
+ }
+
+ public static void initTableMRJob(Job job, String intable, String outtable, String auths) throws AccumuloSecurityException {
+ Configuration conf = job.getConfiguration();
+ String username = conf.get(USERNAME);
+ String password = conf.get(PASSWORD);
+ String instance = conf.get(INSTANCE);
+ String zookeepers = conf.get(ZOOKEEPERS);
+
+ System.out.println("Zookeepers are " + auths);
+
+ if (zookeepers != null) {
+ AccumuloInputFormat.setZooKeeperInstance(job, instance, zookeepers);
+ AccumuloOutputFormat.setZooKeeperInstance(job, instance, zookeepers);
+ } else {
+ throw new IllegalArgumentException("Must specify either mock or zookeepers");
+ }
+
+ AccumuloInputFormat.setConnectorInfo(job, username, new PasswordToken(password));
+ AccumuloInputFormat.setScanAuthorizations(job, new Authorizations(auths));
+ AccumuloInputFormat.setInputTableName(job, intable);
+ job.setInputFormatClass(AccumuloInputFormat.class);
+ job.setMapOutputKeyClass(Text.class);
+ job.setMapOutputValueClass(IntWritable.class);
+
+ // OUTPUT
+ AccumuloOutputFormat.setConnectorInfo(job, username, new PasswordToken(password));
+ AccumuloOutputFormat.setDefaultTableName(job, outtable);
+ job.setOutputFormatClass(AccumuloOutputFormat.class);
+ job.setOutputKeyClass(Text.class);
+ job.setOutputValueClass(Mutation.class);
+
+ }
+
+ public static void initTabToSeqFileJob(Job job, String intable, String outpath, String auths) throws AccumuloSecurityException {
+
+ Configuration conf = job.getConfiguration();
+ String username = conf.get(USERNAME);
+ String password = conf.get(PASSWORD);
+ String instance = conf.get(INSTANCE);
+ String zookeepers = conf.get(ZOOKEEPERS);
+
+ System.out.println("Zookeepers are " + auths);
+
+ if (zookeepers != null) {
+ AccumuloInputFormat.setZooKeeperInstance(job, instance, zookeepers);
+ } else {
+ throw new IllegalArgumentException("Must specify either mock or zookeepers");
+ }
+
+ AccumuloInputFormat.setConnectorInfo(job, username, new PasswordToken(password));
+ AccumuloInputFormat.setScanAuthorizations(job, new Authorizations(auths));
+ AccumuloInputFormat.setInputTableName(job, intable);
+ job.setInputFormatClass(AccumuloInputFormat.class);
+ job.setMapOutputKeyClass(CompositeType.class);
+ job.setMapOutputValueClass(TripleCard.class);
+
+ // OUTPUT
+ SequenceFileOutputFormat.setOutputPath(job, new Path(outpath));
+ job.setOutputFormatClass(SequenceFileOutputFormat.class);
+ job.setOutputKeyClass(CompositeType.class);
+ job.setOutputValueClass(TripleCard.class);
+
+ }
+
+ public static void initJoinMRJob(Job job, String prospectsPath, String spoPath, Class<? extends Mapper<CompositeType,TripleCard,?,?>> mapperClass,
+ String outPath, String auths) throws AccumuloSecurityException {
+
+ MultipleInputs.addInputPath(job, new Path(prospectsPath), SequenceFileInputFormat.class, mapperClass);
+ MultipleInputs.addInputPath(job, new Path(spoPath), SequenceFileInputFormat.class, mapperClass);
+ job.setMapOutputKeyClass(CompositeType.class);
+ job.setMapOutputValueClass(TripleCard.class);
+
+ SequenceFileOutputFormat.setOutputPath(job, new Path(outPath));
+ job.setOutputFormatClass(SequenceFileOutputFormat.class);
+ job.setOutputKeyClass(TripleEntry.class);
+ job.setOutputValueClass(CardList.class);
+
+ }
+
+ public static Mutation createMutation(TripleRow tripleRow) {
+ Mutation mutation = new Mutation(new Text(tripleRow.getRow()));
+ byte[] columnVisibility = tripleRow.getColumnVisibility();
+ ColumnVisibility cv = columnVisibility == null ? EMPTY_CV : new ColumnVisibility(columnVisibility);
+ Long timestamp = tripleRow.getTimestamp();
+ boolean hasts = timestamp != null;
+ timestamp = timestamp == null ? 0l : timestamp;
+ byte[] value = tripleRow.getValue();
+ Value v = value == null ? EMPTY_VALUE : new Value(value);
+ byte[] columnQualifier = tripleRow.getColumnQualifier();
+ Text cqText = columnQualifier == null ? EMPTY_TEXT : new Text(columnQualifier);
+ byte[] columnFamily = tripleRow.getColumnFamily();
+ Text cfText = columnFamily == null ? EMPTY_TEXT : new Text(columnFamily);
+
+ if (hasts) {
+ mutation.put(cfText, cqText, cv, timestamp, v);
+ } else {
+ mutation.put(cfText, cqText, cv, v);
+
+ }
+ return mutation;
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/extras/rya.prospector/src/main/java/mvm/rya/joinselect/mr/utils/TripleCard.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/java/mvm/rya/joinselect/mr/utils/TripleCard.java b/extras/rya.prospector/src/main/java/mvm/rya/joinselect/mr/utils/TripleCard.java
new file mode 100644
index 0000000..3843600
--- /dev/null
+++ b/extras/rya.prospector/src/main/java/mvm/rya/joinselect/mr/utils/TripleCard.java
@@ -0,0 +1,144 @@
+package mvm.rya.joinselect.mr.utils;
+
+/*
+ * #%L
+ * mvm.rya.rya.prospector
+ * %%
+ * Copyright (C) 2014 - 2015 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.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.io.WritableComparable;
+
+public class TripleCard implements WritableComparable<TripleCard> {
+
+ private CardinalityType card = null;
+ private TripleEntry te = null;
+
+ private CardinalityType tempCard = new CardinalityType();
+ private TripleEntry tempTe = new TripleEntry();
+
+ public TripleCard() {}
+
+ public TripleCard(CardinalityType card) {
+ this.setCard(card);
+ }
+
+ public TripleCard(TripleEntry te) {
+ this.setTE(te);
+ }
+
+ public void setCard(CardinalityType card) {
+ tempCard.set(card);
+ this.card = tempCard;
+ this.te = null;
+ }
+
+ public void setTE(TripleEntry te) {
+ tempTe.setTE(te);
+ this.te = tempTe;
+ this.card = null;
+ }
+
+ public CardinalityType getCard() {
+ return this.card;
+ }
+
+ public TripleEntry getTE() {
+ return this.te;
+ }
+
+ public boolean isCardNull() {
+ return (card == null);
+ }
+
+ public boolean isTeNull() {
+ return (te == null);
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ if (card != null) {
+ out.writeBoolean(true);
+ card.write(out);
+ } else {
+ out.writeBoolean(false);
+ te.write(out);
+ }
+
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ if (in.readBoolean()) {
+ tempCard.readFields(in);
+ card = tempCard;
+ te = null;
+ } else {
+ tempTe.readFields(in);
+ te = tempTe;
+ card = null;
+ }
+ }
+
+ @Override
+ public int hashCode() {
+ int result = 7;
+ if (card != null) {
+ result = result * 17 + card.hashCode();
+ } else {
+ result = result * 17 + te.hashCode();
+ }
+ return result;
+
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o instanceof TripleCard) {
+ TripleCard comp = (TripleCard) o;
+ if (card != null) {
+ return card.equals(comp.card);
+ } else {
+ return te.equals(comp.te);
+ }
+ }
+ return false;
+ }
+
+ @Override
+ public String toString() {
+ if (card != null) {
+ return card.toString();
+ } else {
+ return te.toString();
+ }
+ }
+
+ @Override
+ public int compareTo(TripleCard o) {
+
+ if (card != null) {
+ return card.compareTo(o.card);
+ } else {
+ return te.compareTo(o.te);
+ }
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/extras/rya.prospector/src/main/java/mvm/rya/joinselect/mr/utils/TripleEntry.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/java/mvm/rya/joinselect/mr/utils/TripleEntry.java b/extras/rya.prospector/src/main/java/mvm/rya/joinselect/mr/utils/TripleEntry.java
new file mode 100644
index 0000000..8f3769c
--- /dev/null
+++ b/extras/rya.prospector/src/main/java/mvm/rya/joinselect/mr/utils/TripleEntry.java
@@ -0,0 +1,179 @@
+package mvm.rya.joinselect.mr.utils;
+
+/*
+ * #%L
+ * mvm.rya.rya.prospector
+ * %%
+ * Copyright (C) 2014 - 2015 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.io.DataInput;
+import java.io.DataOutput;
+import java.io.IOException;
+
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.WritableComparable;
+
+public class TripleEntry implements WritableComparable<TripleEntry> {
+
+ private Text first;
+ private Text second;
+ private Text firstPos;
+ private Text secondPos;
+ private Text keyPos;
+
+ public TripleEntry() {
+
+ first = new Text();
+ second = new Text();
+ firstPos = new Text();
+ secondPos = new Text();
+ keyPos = new Text();
+
+ }
+
+ public TripleEntry(String first, String second, String firstPos, String secondPos, String keyPos) {
+ this.first = new Text(first);
+ this.second = new Text(second);
+ this.firstPos = new Text(firstPos);
+ this.secondPos = new Text(secondPos);
+ this.keyPos = new Text(keyPos);
+ }
+
+ public TripleEntry(Text first, Text second, Text firstPos, Text secondPos, Text keyPos) {
+ this.first = first;
+ this.second = second;
+ this.firstPos = firstPos;
+ this.secondPos = secondPos;
+ this.keyPos = keyPos;
+ }
+
+ public void setEntry(Text first, Text second) {
+ this.first = first;
+ this.second = second;
+ }
+
+ public void setPosition(Text firstPos, Text secondPos, Text keyPos) {
+ this.firstPos = firstPos;
+ this.secondPos = secondPos;
+ this.keyPos = keyPos;
+ }
+
+ public void setTE(TripleEntry te) {
+
+ this.first.set(te.first);
+ this.second.set(te.second);
+ this.firstPos.set(te.firstPos);
+ this.secondPos.set(te.secondPos);
+ this.keyPos.set(te.keyPos);
+
+ }
+
+ public Text getFirst() {
+ return this.first;
+ }
+
+ public Text getSecond() {
+ return this.second;
+ }
+
+ public Text getFirstPos() {
+ return this.firstPos;
+ }
+
+ public Text getSecondPos() {
+ return this.secondPos;
+ }
+
+ public Text getKeyPos() {
+ return this.keyPos;
+ }
+
+ @Override
+ public void write(DataOutput out) throws IOException {
+ first.write(out);
+ second.write(out);
+ firstPos.write(out);
+ secondPos.write(out);
+ keyPos.write(out);
+
+ }
+
+ @Override
+ public void readFields(DataInput in) throws IOException {
+ first.readFields(in);
+ second.readFields(in);
+ firstPos.readFields(in);
+ secondPos.readFields(in);
+ keyPos.readFields(in);
+
+ }
+
+ @Override
+ public int hashCode() {
+ int result = 7;
+ result = result * 17 + first.hashCode();
+ result = result * 17 + second.hashCode();
+ result = result * 17 + firstPos.hashCode();
+ result = result * 17 + secondPos.hashCode();
+ result = result * 17 + keyPos.hashCode();
+
+ return result;
+
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (o instanceof TripleEntry) {
+ TripleEntry trip = (TripleEntry) o;
+ return first.equals(trip.first) && second.equals(trip.second) && firstPos.equals(trip.firstPos) && secondPos.equals(trip.secondPos)
+ && keyPos.equals(trip.keyPos);
+
+ }
+ return false;
+ }
+
+ @Override
+ public String toString() {
+ return first + "\t" + firstPos + "\t" + second + "\t" + secondPos + "\t" + keyPos;
+
+ }
+
+ @Override
+ public int compareTo(TripleEntry o) {
+
+ int cmp = first.compareTo(o.first);
+ if (cmp != 0) {
+ return cmp;
+ }
+ cmp = firstPos.compareTo(o.firstPos);
+ if (cmp != 0) {
+ return cmp;
+ }
+ cmp = second.compareTo(o.second);
+ if (cmp != 0) {
+ return cmp;
+ }
+
+ cmp = secondPos.compareTo(o.secondPos);
+ if (cmp != 0) {
+ return cmp;
+ }
+ return keyPos.compareTo(o.keyPos);
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/extras/rya.prospector/src/main/resources/META-INF/services/mvm.rya.prospector.plans.IndexWorkPlan
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/main/resources/META-INF/services/mvm.rya.prospector.plans.IndexWorkPlan b/extras/rya.prospector/src/main/resources/META-INF/services/mvm.rya.prospector.plans.IndexWorkPlan
new file mode 100644
index 0000000..38258c1
--- /dev/null
+++ b/extras/rya.prospector/src/main/resources/META-INF/services/mvm.rya.prospector.plans.IndexWorkPlan
@@ -0,0 +1 @@
+mvm.rya.prospector.plans.impl.CountPlan
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/extras/rya.prospector/src/test/groovy/mvm/rya/prospector/mr/ProspectorTest.groovy
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/test/groovy/mvm/rya/prospector/mr/ProspectorTest.groovy b/extras/rya.prospector/src/test/groovy/mvm/rya/prospector/mr/ProspectorTest.groovy
new file mode 100644
index 0000000..f3ef96e
--- /dev/null
+++ b/extras/rya.prospector/src/test/groovy/mvm/rya/prospector/mr/ProspectorTest.groovy
@@ -0,0 +1,178 @@
+package mvm.rya.prospector.mr
+
+import com.google.common.collect.Iterators
+import com.google.common.collect.Lists
+import mvm.rya.accumulo.AccumuloRyaDAO
+import mvm.rya.accumulo.AccumuloRdfConfiguration
+import mvm.rya.api.persist.RdfEvalStatsDAO
+import mvm.rya.api.domain.RyaStatement
+import mvm.rya.api.domain.RyaType
+import mvm.rya.api.domain.RyaURI
+import mvm.rya.prospector.domain.IndexEntry
+import mvm.rya.prospector.domain.TripleValueType
+import mvm.rya.prospector.service.ProspectorService
+import mvm.rya.prospector.service.ProspectorServiceEvalStatsDAO
+import mvm.rya.prospector.utils.ProspectorConstants
+import org.apache.accumulo.core.client.Instance
+import org.apache.accumulo.core.client.mock.MockInstance
+import org.apache.accumulo.core.security.Authorizations
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.util.ToolRunner
+import org.junit.Test
+import org.openrdf.model.vocabulary.XMLSchema
+import org.openrdf.model.impl.URIImpl
+
+import static org.junit.Assert.assertEquals
+import org.openrdf.model.impl.LiteralImpl
+import org.openrdf.model.Value
+
+/**
+ * Date: 12/4/12
+ * Time: 4:33 PM
+ */
+class ProspectorTest {
+
+ @Test
+ public void testCount() throws Exception {
+
+ Instance mock = new MockInstance("accumulo");
+
+/*
+ * #%L
+ * mvm.rya.rya.prospector
+ * %%
+ * 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%
+ */
+ def connector = mock.getConnector("user", "pass".bytes)
+ def intable = "rya_spo"
+ def outtable = "rya_prospects"
+ if (connector.tableOperations().exists(outtable))
+ connector.tableOperations().delete(outtable)
+ connector.tableOperations().create(outtable)
+
+ AccumuloRyaDAO ryaDAO = new AccumuloRyaDAO();
+ ryaDAO.setConnector(connector);
+ ryaDAO.init()
+
+ ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1234"), new RyaURI("urn:gem#pred"), new RyaType("mydata1")))
+ ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1234"), new RyaURI("urn:gem#pred"), new RyaType("mydata2")))
+ ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1234"), new RyaURI("urn:gem#pred"), new RyaType("12")))
+ ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1235"), new RyaURI("urn:gem#pred"), new RyaType(XMLSchema.INTEGER, "12")))
+ ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1235"), new RyaURI("urn:gem#pred1"), new RyaType("12")))
+
+ def confFile = "stats_cluster_config.xml"
+ def confPath = new Path(getClass().getClassLoader().getResource(confFile).toString())
+ def args = (String[]) [confPath];
+ ToolRunner.run(new Prospector(), args);
+ debugTable(connector, outtable)
+
+ def scanner = connector.createScanner(outtable, new Authorizations("U", "FOUO"))
+ def iter = scanner.iterator()
+// assertEquals(11, Iterators.size(iter))
+
+ ryaDAO.destroy()
+
+ def conf = new Configuration()
+ conf.addResource(confPath)
+ // debugTable(mrInfo, outtable)
+
+ def service = new ProspectorService(connector, outtable)
+ def auths = (String[]) ["U", "FOUO"]
+ def prospects = service.getProspects(auths)
+ def plist = Lists.newArrayList(prospects)
+ assertEquals(1, plist.size())
+
+ def rdfConf = new AccumuloRdfConfiguration(conf)
+ rdfConf.setAuths("U","FOUO")
+
+ prospects = service.getProspectsInRange(System.currentTimeMillis() - 100000, System.currentTimeMillis() + 10000, auths)
+ plist = Lists.newArrayList(prospects)
+ assertEquals(1, plist.size())
+
+ List<String> queryTerms = new ArrayList<String>();
+ queryTerms.add("urn:gem:etype");
+ def query = service.query(plist, ProspectorConstants.COUNT, TripleValueType.entity.name(), queryTerms, XMLSchema.ANYURI.stringValue(), auths)
+ assertEquals(1, query.size())
+// assertEquals(
+// new IndexEntry(index: ProspectorConstants.COUNT, data: "urn:gem:etype", dataType: XMLSchema.ANYURI.stringValue(),
+// tripleValueType: TripleValueType.entity, visibility: "", count: -1, timestamp: plist.get(0)),
+// query.get(0))
+
+ queryTerms = new ArrayList<String>();
+ queryTerms.add("urn:gem:etype#1234");
+ query = service.query(plist, ProspectorConstants.COUNT, TripleValueType.subject.name(), queryTerms, XMLSchema.ANYURI.stringValue(), auths)
+ assertEquals(1, query.size())
+
+ queryTerms = new ArrayList<String>();
+ queryTerms.add("urn:gem#pred");
+ query = service.query(plist, ProspectorConstants.COUNT, TripleValueType.predicate.name(), queryTerms, XMLSchema.ANYURI.stringValue(), auths)
+ assertEquals(1, query.size())
+ assertEquals(
+ new IndexEntry(index: ProspectorConstants.COUNT, data: "urn:gem#pred", dataType: XMLSchema.ANYURI.stringValue(),
+ tripleValueType: TripleValueType.predicate, visibility: "", count: 4l, timestamp: plist.get(0)),
+ query.get(0))
+
+ queryTerms = new ArrayList<String>();
+ queryTerms.add("mydata1");
+ query = service.query(plist, ProspectorConstants.COUNT, TripleValueType.object.name(), queryTerms, XMLSchema.STRING.stringValue(), auths)
+ assertEquals(1, query.size())
+// assertEquals(
+// new IndexEntry(index: ProspectorConstants.COUNT, data: "mydata1", dataType: XMLSchema.STRING.stringValue(),
+// tripleValueType: TripleValueType.object, visibility: "", count: -1, timestamp: plist.get(0)),
+// query.get(0))
+
+ queryTerms = new ArrayList<String>();
+ queryTerms.add("urn:gem:etype#1234");
+ queryTerms.add("urn:gem#pred");
+ query = service.query(plist, ProspectorConstants.COUNT, TripleValueType.subjectpredicate.name(), queryTerms, XMLSchema.STRING.stringValue(), auths)
+ assertEquals(1, query.size())
+// assertEquals(
+// new IndexEntry(index: ProspectorConstants.COUNT, data: "urn:gem:etype#1234" + "\u0000" + "urn:gem#pred", dataType: XMLSchema.STRING.stringValue(),
+// tripleValueType: TripleValueType.subjectpredicate, visibility: "", count: -1, timestamp: plist.get(0)),
+// query.get(0))
+
+ queryTerms = new ArrayList<String>();
+ queryTerms.add("urn:gem#pred");
+ queryTerms.add("12");
+ query = service.query(plist, ProspectorConstants.COUNT, TripleValueType.predicateobject.name(), queryTerms, XMLSchema.STRING.stringValue(), auths)
+ assertEquals(1, query.size())
+// assertEquals(
+// new IndexEntry(index: ProspectorConstants.COUNT, data: "urn:gem#pred" + "\u0000" + "12", dataType: XMLSchema.STRING.stringValue(),
+// tripleValueType: TripleValueType.predicateobject, visibility: "", count: -1, timestamp: plist.get(0)),
+// query.get(0))
+
+ queryTerms = new ArrayList<String>();
+ queryTerms.add("urn:gem:etype#1234");
+ queryTerms.add("mydata1");
+ query = service.query(plist, ProspectorConstants.COUNT, TripleValueType.subjectobject.name(), queryTerms, XMLSchema.STRING.stringValue(), auths)
+
+ assertEquals(1, query.size())
+// assertEquals(
+// new IndexEntry(index: ProspectorConstants.COUNT, data: "urn:gem:etype#1234" + "\u0000" + "mydata1", dataType: XMLSchema.STRING.stringValue(),
+// tripleValueType: TripleValueType.subjectobject, visibility: "", count: -1, timestamp: plist.get(0)),
+// query.get(0))
+
+ //should be in a teardown method
+ connector.tableOperations().delete(outtable)
+ }
+
+ private void debugTable(def connector, String table) {
+ connector.createScanner(table, new Authorizations((String[]) ["U", "FOUO"])).iterator().each {
+ println it
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/extras/rya.prospector/src/test/groovy/mvm/rya/prospector/service/ProspectorServiceEvalStatsDAOTest.groovy
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/test/groovy/mvm/rya/prospector/service/ProspectorServiceEvalStatsDAOTest.groovy b/extras/rya.prospector/src/test/groovy/mvm/rya/prospector/service/ProspectorServiceEvalStatsDAOTest.groovy
new file mode 100644
index 0000000..275c6d5
--- /dev/null
+++ b/extras/rya.prospector/src/test/groovy/mvm/rya/prospector/service/ProspectorServiceEvalStatsDAOTest.groovy
@@ -0,0 +1,182 @@
+package mvm.rya.prospector.service
+
+import com.google.common.collect.Iterators
+import mvm.rya.accumulo.AccumuloRdfConfiguration
+import mvm.rya.accumulo.AccumuloRyaDAO
+import mvm.rya.api.domain.RyaStatement
+import mvm.rya.api.domain.RyaType
+import mvm.rya.api.domain.RyaURI
+import mvm.rya.api.persist.RdfEvalStatsDAO
+import mvm.rya.prospector.mr.Prospector
+import org.apache.accumulo.core.client.Instance
+import org.apache.accumulo.core.client.mock.MockInstance
+import org.apache.accumulo.core.security.Authorizations
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.fs.Path
+import org.apache.hadoop.util.ToolRunner
+import org.junit.Test
+import org.openrdf.model.impl.URIImpl
+import org.openrdf.model.vocabulary.XMLSchema
+
+import static org.junit.Assert.assertEquals
+import org.openrdf.model.impl.LiteralImpl
+import org.openrdf.model.Value
+
+/**
+ * Date: 1/26/13
+ * Time: 3:00 PM
+ */
+class ProspectorServiceEvalStatsDAOTest {
+
+ @Test
+ public void testCount() throws Exception {
+
+ Instance mock = new MockInstance("accumulo");
+
+/*
+ * #%L
+ * mvm.rya.rya.prospector
+ * %%
+ * 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%
+ */
+ def connector = mock.getConnector("user", "pass".bytes)
+ def intable = "rya_spo"
+ def outtable = "rya_prospects"
+ if (connector.tableOperations().exists(outtable))
+ connector.tableOperations().delete(outtable)
+ connector.tableOperations().create(outtable)
+
+ AccumuloRyaDAO ryaDAO = new AccumuloRyaDAO();
+ ryaDAO.setConnector(connector);
+ ryaDAO.init()
+
+ ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1234"), new RyaURI("urn:gem#pred"), new RyaType("mydata1")))
+ ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1234"), new RyaURI("urn:gem#pred"), new RyaType("mydata2")))
+ ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1234"), new RyaURI("urn:gem#pred"), new RyaType("12")))
+ ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1235"), new RyaURI("urn:gem#pred"), new RyaType(XMLSchema.INTEGER, "12")))
+ ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1235"), new RyaURI("urn:gem#pred1"), new RyaType("12")))
+
+ def confFile = "stats_cluster_config.xml"
+ def confPath = new Path(getClass().getClassLoader().getResource(confFile).toString())
+ def args = (String[]) [confPath];
+ ToolRunner.run(new Prospector(), args);
+ debugTable(connector, outtable)
+
+ def scanner = connector.createScanner(outtable, new Authorizations("U", "FOUO"))
+ def iter = scanner.iterator()
+// assertEquals(11, Iterators.size(iter))
+
+ ryaDAO.destroy()
+
+ def conf = new Configuration()
+ conf.addResource(confPath)
+// debugTable(connector, outtable)
+
+ def rdfConf = new AccumuloRdfConfiguration(conf)
+ rdfConf.setAuths("U","FOUO")
+ def evalDao = new ProspectorServiceEvalStatsDAO(connector, rdfConf)
+ evalDao.init()
+
+ List<Value> values = new ArrayList<Value>();
+ values.add( new URIImpl("urn:gem#pred"));
+
+ def count = evalDao.getCardinality(rdfConf, RdfEvalStatsDAO.CARDINALITY_OF.PREDICATE, values)
+ assertEquals(4.0, count, 0.001);
+
+ values = new ArrayList<Value>();
+ values.add( new LiteralImpl("mydata1"));
+
+ count = evalDao.getCardinality(rdfConf, RdfEvalStatsDAO.CARDINALITY_OF.OBJECT, values);
+ assertEquals(1.0, count, 0.001);
+
+ values = new ArrayList<Value>();
+ values.add( new LiteralImpl("mydata3"));
+
+ count = evalDao.getCardinality(rdfConf, RdfEvalStatsDAO.CARDINALITY_OF.OBJECT, values);
+ assertEquals(-1.0, count, 0.001);
+
+ //should be in a teardown method
+ connector.tableOperations().delete(outtable)
+ }
+
+ @Test
+ public void testNoAuthsCount() throws Exception {
+
+ Instance mock = new MockInstance("accumulo");
+ def connector = mock.getConnector("user", "pass".bytes)
+ def intable = "rya_spo"
+ def outtable = "rya_prospects"
+ if (connector.tableOperations().exists(outtable))
+ connector.tableOperations().delete(outtable)
+ connector.tableOperations().create(outtable)
+ connector.securityOperations().createUser("user", "pass".bytes, new Authorizations("U", "FOUO"))
+
+ AccumuloRyaDAO ryaDAO = new AccumuloRyaDAO();
+ ryaDAO.setConnector(connector);
+ ryaDAO.init()
+
+ ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1234"), new RyaURI("urn:gem#pred"), new RyaType("mydata1")))
+ ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1234"), new RyaURI("urn:gem#pred"), new RyaType("mydata2")))
+ ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1234"), new RyaURI("urn:gem#pred"), new RyaType("12")))
+ ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1235"), new RyaURI("urn:gem#pred"), new RyaType(XMLSchema.INTEGER, "12")))
+ ryaDAO.add(new RyaStatement(new RyaURI("urn:gem:etype#1235"), new RyaURI("urn:gem#pred1"), new RyaType("12")))
+
+ def confFile = "stats_cluster_config.xml"
+ def confPath = new Path(getClass().getClassLoader().getResource(confFile).toString())
+ def args = (String[]) [confPath];
+ ToolRunner.run(new Prospector(), args);
+
+ def scanner = connector.createScanner(outtable, new Authorizations("U", "FOUO"))
+ def iter = scanner.iterator()
+// assertEquals(11, Iterators.size(iter))
+
+ ryaDAO.destroy()
+
+ def conf = new Configuration()
+ conf.addResource(confPath)
+
+ def rdfConf = new AccumuloRdfConfiguration(conf)
+// rdfConf.setAuths("U","FOUO")
+ def evalDao = new ProspectorServiceEvalStatsDAO(connector, rdfConf)
+ evalDao.init()
+
+
+ List<Value> values = new ArrayList<Value>();
+ values.add( new URIImpl("urn:gem#pred"));
+ def count = evalDao.getCardinality(rdfConf, RdfEvalStatsDAO.CARDINALITY_OF.PREDICATE, values)
+ assertEquals(4.0, count, 0.001);
+
+ values = new ArrayList<Value>();
+ values.add( new LiteralImpl("mydata1"));
+ count = evalDao.getCardinality(rdfConf, RdfEvalStatsDAO.CARDINALITY_OF.OBJECT, values);
+ assertEquals(1.0, count, 0.001);
+
+ values = new ArrayList<Value>();
+ values.add( new LiteralImpl("mydata3"));
+
+ count = evalDao.getCardinality(rdfConf, RdfEvalStatsDAO.CARDINALITY_OF.OBJECT, values);
+ assertEquals(-1.0, count, 0.001);
+
+ //should be in a teardown method
+ connector.tableOperations().delete(outtable)
+ }
+
+ private void debugTable(def connector, String table) {
+ connector.createScanner(table, new Authorizations((String[]) ["U", "FOUO"])).iterator().each {
+ println it
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/extras/rya.prospector/src/test/java/mvm/rya/joinselect/AccumuloSelectivityEvalDAOTest.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/test/java/mvm/rya/joinselect/AccumuloSelectivityEvalDAOTest.java b/extras/rya.prospector/src/test/java/mvm/rya/joinselect/AccumuloSelectivityEvalDAOTest.java
new file mode 100644
index 0000000..007af96
--- /dev/null
+++ b/extras/rya.prospector/src/test/java/mvm/rya/joinselect/AccumuloSelectivityEvalDAOTest.java
@@ -0,0 +1,591 @@
+package mvm.rya.joinselect;
+
+/*
+ * #%L
+ * mvm.rya.rya.prospector
+ * %%
+ * Copyright (C) 2014 - 2015 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.math.BigDecimal;
+import java.math.MathContext;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import mvm.rya.accumulo.AccumuloRdfConfiguration;
+import mvm.rya.api.RdfCloudTripleStoreConfiguration;
+import mvm.rya.api.layout.TablePrefixLayoutStrategy;
+import mvm.rya.api.persist.RdfEvalStatsDAO;
+import mvm.rya.joinselect.AccumuloSelectivityEvalDAO;
+import mvm.rya.prospector.service.ProspectorServiceEvalStatsDAO;
+
+import org.apache.accumulo.core.client.AccumuloException;
+import org.apache.accumulo.core.client.AccumuloSecurityException;
+import org.apache.accumulo.core.client.BatchWriter;
+import org.apache.accumulo.core.client.BatchWriterConfig;
+import org.apache.accumulo.core.client.Connector;
+import org.apache.accumulo.core.client.Instance;
+import org.apache.accumulo.core.client.Scanner;
+import org.apache.accumulo.core.client.TableExistsException;
+import org.apache.accumulo.core.client.TableNotFoundException;
+import org.apache.accumulo.core.client.admin.TableOperations;
+import org.apache.accumulo.core.client.mock.MockInstance;
+import org.apache.accumulo.core.client.security.tokens.PasswordToken;
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Range;
+import org.apache.accumulo.core.data.Value;
+import org.apache.accumulo.core.security.Authorizations;
+import org.apache.hadoop.io.Text;
+import org.junit.Assert;
+import org.junit.Before;
+import org.junit.Test;
+import org.openrdf.query.MalformedQueryException;
+import org.openrdf.query.algebra.StatementPattern;
+import org.openrdf.query.algebra.TupleExpr;
+import org.openrdf.query.algebra.helpers.StatementPatternCollector;
+import org.openrdf.query.parser.ParsedQuery;
+import org.openrdf.query.parser.sparql.SPARQLParser;
+
+import com.google.common.collect.Lists;
+
+public class AccumuloSelectivityEvalDAOTest {
+
+ private static final String DELIM = "\u0000";
+ private final byte[] EMPTY_BYTE = new byte[0];
+ private final Value EMPTY_VAL = new Value(EMPTY_BYTE);
+
+ private String q1 = ""//
+ + "SELECT ?h " //
+ + "{" //
+ + " ?h <uri:howlsAt> <uri:moon>. "//
+ + " ?h <http://www.w3.org/2000/01/rdf-schema#label> <uri:dog> ."//
+ + " ?h <uri:barksAt> <uri:cat> ."//
+ + " ?h <uri:peesOn> <uri:hydrant> . "//
+ + "}";//
+
+ private String q2 = ""//
+ + "SELECT ?h " //
+ + "{" //
+ + " <uri:howlsAt> ?h <uri:moon>. "//
+ + " <http://www.w3.org/2000/01/rdf-schema#label> ?h <uri:dog> ."//
+ + " <uri:barksAt> ?h <uri:cat> ."//
+ + " <uri:peesOn> ?h <uri:hydrant> . "//
+ + "}";//
+
+ private String q3 = ""//
+ + "SELECT ?h " //
+ + "{" //
+ + " <uri:howlsAt> <uri:moon> ?h. "//
+ + " <http://www.w3.org/2000/01/rdf-schema#label> <uri:dog> ?h ."//
+ + " <uri:barksAt> ?h <uri:cat> ."//
+ + " ?h <uri:peesOn> <uri:hydrant> . "//
+ + "}";//
+
+ private Connector conn;
+ AccumuloRdfConfiguration arc;
+ RdfEvalStatsDAO<RdfCloudTripleStoreConfiguration> res;
+ BatchWriterConfig config;
+ Instance mock;
+
+ @Before
+ public void init() throws AccumuloException, AccumuloSecurityException, TableNotFoundException, TableExistsException {
+
+ mock = new MockInstance("accumulo");
+ PasswordToken pToken = new PasswordToken("pass".getBytes());
+ conn = mock.getConnector("user", pToken);
+
+ config = new BatchWriterConfig();
+ config.setMaxMemory(1000);
+ config.setMaxLatency(1000, TimeUnit.SECONDS);
+ config.setMaxWriteThreads(10);
+
+ if (conn.tableOperations().exists("rya_prospects")) {
+ conn.tableOperations().delete("rya_prospects");
+ }
+ if (conn.tableOperations().exists("rya_selectivity")) {
+ conn.tableOperations().delete("rya_selectivity");
+ }
+
+ arc = new AccumuloRdfConfiguration();
+ res = new ProspectorServiceEvalStatsDAO(conn, arc);
+ arc.setTableLayoutStrategy(new TablePrefixLayoutStrategy());
+ arc.setMaxRangesForScanner(300);
+
+ }
+
+ @Test
+ public void testInitialize() throws AccumuloException, AccumuloSecurityException, TableNotFoundException, TableExistsException {
+
+ AccumuloSelectivityEvalDAO accc = new AccumuloSelectivityEvalDAO();
+ accc.setConf(arc);
+ accc.setConnector(conn);
+ accc.setRdfEvalDAO(res);
+ accc.init();
+
+ TableOperations tos = conn.tableOperations();
+ Assert.assertTrue(tos.exists("rya_prospects") && tos.exists("rya_selectivity"));
+ Assert.assertTrue(accc.isInitialized());
+ Assert.assertTrue(accc.getConf().equals(arc));
+ Assert.assertTrue(accc.getConnector().equals(conn));
+ Assert.assertTrue(accc.getRdfEvalDAO().equals(res));
+
+ }
+
+ @Test
+ public void testCardinalityQuery1() throws AccumuloException, AccumuloSecurityException, TableExistsException, TableNotFoundException,
+ MalformedQueryException {
+
+ AccumuloSelectivityEvalDAO accc = new AccumuloSelectivityEvalDAO();
+ accc.setConf(arc);
+ accc.setRdfEvalDAO(res);
+ accc.setConnector(conn);
+ accc.init();
+
+ BatchWriter bw = conn.createBatchWriter("rya_prospects", config);
+
+ BatchWriter bw1 = conn.createBatchWriter("rya_selectivity", config);
+ Mutation m = new Mutation(new Text("subjectpredicateobject" + DELIM + "FullTableCardinality"));
+ m.put(new Text("FullTableCardinality"), new Text("600"), EMPTY_VAL);
+ List<Mutation> list = Lists.newArrayList();
+ list.add(m);
+ bw1.addMutations(list);
+ bw1.close();
+
+ String s1 = "predicateobject" + DELIM + "http://www.w3.org/2000/01/rdf-schema#label" + DELIM + "uri:dog";
+ String s2 = "predicateobject" + DELIM + "uri:barksAt" + DELIM + "uri:cat";
+ String s3 = "predicateobject" + DELIM + "uri:peesOn" + DELIM + "uri:hydrant";
+ List<Mutation> mList = new ArrayList<Mutation>();
+ Mutation m1, m2, m3;
+
+ Integer tempInt;
+ Integer tempInt2;
+
+ for (int i = 1; i < 7; i++) {
+ tempInt = 5 * i;
+ tempInt2 = 10 - i;
+ m1 = new Mutation(s1 + DELIM + i);
+ m1.put(new Text("count"), new Text(""), new Value((tempInt.toString()).getBytes()));
+ m2 = new Mutation(s2 + DELIM + (7 - i));
+ m2.put(new Text("count"), new Text(""), new Value((tempInt.toString()).getBytes()));
+ m3 = new Mutation(s3 + DELIM + (10 + i));
+ m3.put(new Text("count"), new Text(""), new Value((tempInt2.toString()).getBytes()));
+ mList.add(m1);
+ mList.add(m2);
+ mList.add(m3);
+ }
+
+ bw.addMutations(mList);
+ bw.close();
+
+ List<StatementPattern> spList = getSpList(q1);
+ long c1 = accc.getCardinality(arc, spList.get(0));
+ long c2 = accc.getCardinality(arc, spList.get(1));
+ long c3 = accc.getCardinality(arc, spList.get(2));
+ long c4 = accc.getCardinality(arc, spList.get(3));
+
+ Assert.assertTrue(c1 == (long) 0);
+ Assert.assertTrue(c2 == (long) 5);
+ Assert.assertTrue(c3 == (long) 30);
+ Assert.assertTrue(c4 == (long) 9);
+
+ }
+
+ @Test
+ public void testCardinalityQuery2() throws AccumuloException, AccumuloSecurityException, TableExistsException, TableNotFoundException,
+ MalformedQueryException {
+
+ AccumuloSelectivityEvalDAO accc = new AccumuloSelectivityEvalDAO();
+ accc.setConf(arc);
+ accc.setConnector(conn);
+ accc.setRdfEvalDAO(res);
+ accc.init();
+
+ BatchWriter bw = conn.createBatchWriter("rya_prospects", config);
+
+ BatchWriter bw1 = conn.createBatchWriter("rya_selectivity", config);
+ Mutation m = new Mutation(new Text("subjectpredicateobject" + DELIM + "FullTableCardinality"));
+ m.put(new Text("FullTableCardinality"), new Text("600"), EMPTY_VAL);
+ List<Mutation> list = Lists.newArrayList();
+ list.add(m);
+ bw1.addMutations(list);
+ bw1.close();
+
+ String s1 = "subjectobject" + DELIM + "http://www.w3.org/2000/01/rdf-schema#label" + DELIM + "uri:dog";
+ String s2 = "subjectobject" + DELIM + "uri:barksAt" + DELIM + "uri:cat";
+ String s3 = "subjectobject" + DELIM + "uri:peesOn" + DELIM + "uri:hydrant";
+ List<Mutation> mList = new ArrayList<Mutation>();
+ Mutation m1, m2, m3;
+
+ Integer tempInt;
+ Integer tempInt2;
+
+ for (int i = 1; i < 7; i++) {
+ tempInt = 5 * i;
+ tempInt2 = 10 - i;
+ m1 = new Mutation(s1 + DELIM + i);
+ m1.put(new Text("count"), new Text(""), new Value((tempInt.toString()).getBytes()));
+ m2 = new Mutation(s2 + DELIM + (7 - i));
+ m2.put(new Text("count"), new Text(""), new Value((tempInt.toString()).getBytes()));
+ m3 = new Mutation(s3 + DELIM + (10 + i));
+ m3.put(new Text("count"), new Text(""), new Value((tempInt2.toString()).getBytes()));
+ mList.add(m1);
+ mList.add(m2);
+ mList.add(m3);
+ }
+ bw.addMutations(mList);
+ bw.close();
+
+ List<StatementPattern> spList = getSpList(q2);
+ long c1 = accc.getCardinality(arc, spList.get(0));
+ long c2 = accc.getCardinality(arc, spList.get(1));
+ long c3 = accc.getCardinality(arc, spList.get(2));
+ long c4 = accc.getCardinality(arc, spList.get(3));
+
+ Assert.assertTrue(c1 == (long) 0);
+ Assert.assertTrue(c2 == (long) 5);
+ Assert.assertTrue(c3 == (long) 30);
+ Assert.assertTrue(c4 == (long) 9);
+
+ }
+
+ @Test
+ public void testJoinCardinalityQuery1() throws AccumuloException, AccumuloSecurityException, TableExistsException,
+ TableNotFoundException, MalformedQueryException {
+
+ AccumuloSelectivityEvalDAO accc = new AccumuloSelectivityEvalDAO();
+ accc.setConf(arc);
+ accc.setConnector(conn);
+ accc.setRdfEvalDAO(res);
+ accc.init();
+
+ BatchWriter bw1 = conn.createBatchWriter("rya_prospects", config);
+ BatchWriter bw2 = conn.createBatchWriter("rya_selectivity", config);
+
+ String s1 = "predicateobject" + DELIM + "http://www.w3.org/2000/01/rdf-schema#label" + DELIM + "uri:dog";
+ String s2 = "predicateobject" + DELIM + "uri:barksAt" + DELIM + "uri:cat";
+ String s3 = "predicateobject" + DELIM + "uri:peesOn" + DELIM + "uri:hydrant";
+ List<Mutation> mList = new ArrayList<Mutation>();
+ List<Mutation> mList2 = new ArrayList<Mutation>();
+ List<String> sList = Arrays.asList("subjectobject", "subjectpredicate", "subjectsubject", "predicateobject", "predicatepredicate",
+ "predicatesubject");
+ Mutation m1, m2, m3, m4;
+
+ m1 = new Mutation(s1 + DELIM + "1");
+ m1.put(new Text("count"), new Text(""), new Value("20".getBytes()));
+ m2 = new Mutation(s2 + DELIM + "2");
+ m2.put(new Text("count"), new Text(""), new Value("15".getBytes()));
+ m3 = new Mutation(s3 + DELIM + "3");
+ m3.put(new Text("count"), new Text(""), new Value("10".getBytes()));
+ mList.add(m1);
+ mList.add(m2);
+ mList.add(m3);
+
+ bw1.addMutations(mList);
+ bw1.close();
+
+ m1 = new Mutation(s1);
+ m2 = new Mutation(s2);
+ m3 = new Mutation(s3);
+ int i = 30;
+ int j = 60;
+ int k = 90;
+ Long count1;
+ Long count2;
+ Long count3;
+
+ for (String s : sList) {
+ count1 = (long) i;
+ count2 = (long) j;
+ count3 = (long) k;
+ m1.put(new Text(s), new Text(count1.toString()), EMPTY_VAL);
+ m2.put(new Text(s), new Text(count2.toString()), EMPTY_VAL);
+ m3.put(new Text(s), new Text(count3.toString()), EMPTY_VAL);
+ i = 2 * i;
+ j = 2 * j;
+ k = 2 * k;
+ }
+ m4 = new Mutation(new Text("subjectpredicateobject" + DELIM + "FullTableCardinality"));
+ m4.put(new Text("FullTableCardinality"), new Text("600"), EMPTY_VAL);
+ mList2.add(m1);
+ mList2.add(m2);
+ mList2.add(m3);
+ mList2.add(m4);
+ bw2.addMutations(mList2);
+ bw2.close();
+
+ Scanner scan = conn.createScanner("rya_selectivity", new Authorizations());
+ scan.setRange(new Range());
+
+ for (Map.Entry<Key, Value> entry : scan) {
+ System.out.println("Key row string is " + entry.getKey().getRow().toString());
+ System.out.println("Key is " + entry.getKey());
+ System.out.println("Value is " + (new String(entry.getKey().getColumnQualifier().toString())));
+
+ }
+
+ List<StatementPattern> spList = getSpList(q1);
+ System.out.println(spList);
+ List<Double> jCardList = new ArrayList<Double>();
+
+ for (StatementPattern sp1 : spList) {
+ for (StatementPattern sp2 : spList) {
+ jCardList.add(accc.getJoinSelect(arc, sp1, sp2));
+ }
+ }
+
+ System.out.println("Join cardinalities are " + jCardList);
+
+ Assert.assertEquals(0, jCardList.get(0), .001);
+ Assert.assertEquals(0, jCardList.get(3), .001);
+ Assert.assertEquals(6.0 / 600, jCardList.get(5), .001);
+ Assert.assertEquals(6.0 / 600, jCardList.get(6), .001);
+ Assert.assertEquals(0 / 600, jCardList.get(8), .001);
+ Assert.assertEquals(6.0 / 600, jCardList.get(7), .001);
+ Assert.assertEquals(15.0 / 600, jCardList.get(11), .001);
+ Assert.assertEquals(6.0 / 600, jCardList.get(13), .001);
+ Assert.assertEquals(10.0 / 600, jCardList.get(15), .001);
+
+ Assert.assertTrue(jCardList.get(0) == 0);
+ Assert.assertTrue(jCardList.get(3) == 0);
+ Assert.assertTrue(jCardList.get(5) == .01);
+ Assert.assertTrue(jCardList.get(6) == .01);
+ Assert.assertTrue(jCardList.get(8) == 0);
+ Assert.assertTrue(jCardList.get(7) == (6.0 / 600));
+ Assert.assertTrue(jCardList.get(11) == (1.0 / 40));
+ Assert.assertTrue(jCardList.get(13) == .01);
+ Assert.assertTrue(jCardList.get(15) == (10.0 / 600));
+
+ }
+
+ @Test
+ public void testJoinCardinalityQuery2() throws AccumuloException, AccumuloSecurityException, TableExistsException,
+ TableNotFoundException, MalformedQueryException {
+
+ AccumuloSelectivityEvalDAO accc = new AccumuloSelectivityEvalDAO();
+ accc.setConf(arc);
+ accc.setConnector(conn);
+ accc.setRdfEvalDAO(res);
+ accc.init();
+
+ BatchWriter bw1 = conn.createBatchWriter("rya_prospects", config);
+ BatchWriter bw2 = conn.createBatchWriter("rya_selectivity", config);
+
+ String s1 = "subjectobject" + DELIM + "http://www.w3.org/2000/01/rdf-schema#label" + DELIM + "uri:dog";
+ String s2 = "subjectobject" + DELIM + "uri:barksAt" + DELIM + "uri:cat";
+ String s3 = "subjectobject" + DELIM + "uri:peesOn" + DELIM + "uri:hydrant";
+ String s4 = "objectsubject" + DELIM + "uri:dog" + DELIM + "http://www.w3.org/2000/01/rdf-schema#label";
+ String s5 = "objectsubject" + DELIM + "uri:cat" + DELIM + "uri:barksAt";
+ String s6 = "objectsubject" + DELIM + "uri:hydrant" + DELIM + "uri:peesOn";
+ List<String> sList = Arrays.asList("subjectobject", "subjectpredicate", "subjectsubject", "predicateobject", "predicatepredicate",
+ "predicatesubject");
+ List<Mutation> mList = new ArrayList<Mutation>();
+ List<Mutation> mList2 = new ArrayList<Mutation>();
+ Mutation m1, m2, m3, m4;
+
+ m1 = new Mutation(s1 + DELIM + "1");
+ m1.put(new Text("count"), new Text(""), new Value("2".getBytes()));
+ m2 = new Mutation(s2 + DELIM + "2");
+ m2.put(new Text("count"), new Text(""), new Value("4".getBytes()));
+ m3 = new Mutation(s3 + DELIM + "3");
+ m3.put(new Text("count"), new Text(""), new Value("6".getBytes()));
+ mList.add(m1);
+ mList.add(m2);
+ mList.add(m3);
+
+ bw1.addMutations(mList);
+ bw1.close();
+
+ m1 = new Mutation(s4);
+ m2 = new Mutation(s5);
+ m3 = new Mutation(s6);
+ int i = 5;
+ int j = 6;
+ int k = 7;
+ Long count1;
+ Long count2;
+ Long count3;
+
+ for (String s : sList) {
+ count1 = (long) i;
+ count2 = (long) j;
+ count3 = (long) k;
+ m1.put(new Text(s), new Text(count1.toString()), EMPTY_VAL);
+ m2.put(new Text(s), new Text(count2.toString()), EMPTY_VAL);
+ m3.put(new Text(s), new Text(count3.toString()), EMPTY_VAL);
+ i = 2 * i;
+ j = 2 * j;
+ k = 2 * k;
+ }
+ m4 = new Mutation(new Text("subjectpredicateobject" + DELIM + "FullTableCardinality"));
+ m4.put(new Text("FullTableCardinality"), new Text("600"), EMPTY_VAL);
+ mList2.add(m1);
+ mList2.add(m2);
+ mList2.add(m3);
+ mList2.add(m4);
+ bw2.addMutations(mList2);
+ bw2.close();
+
+ List<StatementPattern> spList = getSpList(q2);
+ // System.out.println(spList);
+ List<Double> jCardList = new ArrayList<Double>();
+
+ for (StatementPattern sp1 : spList) {
+ for (StatementPattern sp2 : spList) {
+ jCardList.add(accc.getJoinSelect(arc, sp1, sp2));
+ }
+ }
+
+ System.out.println("Join cardinalities are " + jCardList);
+
+ Assert.assertEquals(0, jCardList.get(0), .001);
+ Assert.assertEquals(0, jCardList.get(3), .001);
+ Assert.assertEquals(2.0 / 600, jCardList.get(5), .001);
+ Assert.assertEquals(4.0 / 600, jCardList.get(6), .001);
+ Assert.assertEquals(.0 / 600, jCardList.get(8), .001);
+ Assert.assertEquals(6. / 600, jCardList.get(7), .001);
+ Assert.assertEquals(6. / 600, jCardList.get(11), .001);
+
+ Assert.assertTrue(jCardList.get(0) == 0);
+ Assert.assertTrue(jCardList.get(3) == 0);
+ Assert.assertTrue(jCardList.get(5) == (1.0 / 300));
+ Assert.assertTrue(jCardList.get(6) == (4.0 / 600));
+ Assert.assertTrue(jCardList.get(8) == 0);
+ Assert.assertTrue(jCardList.get(7) == .01);
+ Assert.assertTrue(jCardList.get(11) == .01);
+
+ }
+
+ @Test
+ public void testJoinCardinalityQuery3() throws AccumuloException, AccumuloSecurityException, TableExistsException,
+ TableNotFoundException, MalformedQueryException {
+
+ AccumuloSelectivityEvalDAO accc = new AccumuloSelectivityEvalDAO();
+ accc.setConf(arc);
+ accc.setConnector(conn);
+ accc.setRdfEvalDAO(res);
+ accc.init();
+
+ BatchWriter bw1 = conn.createBatchWriter("rya_prospects", config);
+ BatchWriter bw2 = conn.createBatchWriter("rya_selectivity", config);
+
+ String s1 = "subjectpredicate" + DELIM + "http://www.w3.org/2000/01/rdf-schema#label" + DELIM + "uri:dog";
+ String s2 = "subjectobject" + DELIM + "uri:barksAt" + DELIM + "uri:cat";
+ String s3 = "predicateobject" + DELIM + "uri:peesOn" + DELIM + "uri:hydrant";
+ String s4 = "subjectpredicate" + DELIM + "uri:howlsAt" + DELIM + "uri:moon";
+ String s5 = "objectsubject" + DELIM + "uri:cat" + DELIM + "uri:barksAt";
+
+ List<String> sList = Arrays.asList("subjectobject", "objectsubject", "objectobject", "objectpredicate", "subjectpredicate",
+ "subjectsubject", "predicateobject", "predicatepredicate", "predicatesubject");
+ List<Mutation> mList = new ArrayList<Mutation>();
+ List<Mutation> mList2 = new ArrayList<Mutation>();
+ Mutation m1, m2, m3, m4, m5, m6;
+
+ m1 = new Mutation(s1 + DELIM + "1");
+ m1.put(new Text("count"), new Text(""), new Value("15".getBytes()));
+ m2 = new Mutation(s2 + DELIM + "2");
+ m2.put(new Text("count"), new Text(""), new Value("11".getBytes()));
+ m3 = new Mutation(s3 + DELIM + "3");
+ m3.put(new Text("count"), new Text(""), new Value("13".getBytes()));
+ m4 = new Mutation(s4 + DELIM + "8");
+ m4.put(new Text("count"), new Text(""), new Value("20".getBytes()));
+ m5 = new Mutation(s4 + DELIM + "2");
+ m5.put(new Text("count"), new Text(""), new Value("10".getBytes()));
+
+ mList.add(m1);
+ mList.add(m2);
+ mList.add(m3);
+ mList.add(m4);
+ mList.add(m5);
+
+ bw1.addMutations(mList);
+ bw1.close();
+
+ m1 = new Mutation(s1);
+ m2 = new Mutation(s5);
+ m3 = new Mutation(s3);
+ m4 = new Mutation(s4);
+ int i = 5;
+ int j = 6;
+ int k = 7;
+ int l = 8;
+ Long count1;
+ Long count2;
+ Long count3;
+ Long count4;
+
+ for (String s : sList) {
+ count1 = (long) i;
+ count2 = (long) j;
+ count3 = (long) k;
+ count4 = (long) l;
+ m1.put(new Text(s), new Text(count1.toString()), EMPTY_VAL);
+ m2.put(new Text(s), new Text(count2.toString()), EMPTY_VAL);
+ m3.put(new Text(s), new Text(count3.toString()), EMPTY_VAL);
+ m4.put(new Text(s), new Text(count4.toString()), EMPTY_VAL);
+ i = 2 * i;
+ j = 2 * j;
+ k = 2 * k;
+ l = 2 * l;
+ }
+ m5 = new Mutation(new Text("subjectpredicateobject" + DELIM + "FullTableCardinality"));
+ m5.put(new Text("FullTableCardinality"), new Text("600"), EMPTY_VAL);
+ mList2.add(m1);
+ mList2.add(m2);
+ mList2.add(m3);
+ mList2.add(m4);
+ mList2.add(m5);
+ bw2.addMutations(mList2);
+ bw2.close();
+
+ List<StatementPattern> spList = getSpList(q3);
+ System.out.println(spList);
+ List<Double> jCardList = new ArrayList<Double>();
+
+ for (StatementPattern sp1 : spList) {
+ for (StatementPattern sp2 : spList) {
+ jCardList.add(accc.getJoinSelect(arc, sp1, sp2));
+ }
+ }
+
+ MathContext mc = new MathContext(3);
+
+ Assert.assertEquals(3.2 / 600, jCardList.get(0), .001);
+ Assert.assertEquals(0.5384615384615384 / 600, jCardList.get(3), .001);
+ Assert.assertEquals(1.3333333333333333 / 600, jCardList.get(5), .001);
+ Assert.assertEquals(2.6666666666666665 / 600, jCardList.get(6), .001);
+ Assert.assertEquals(6.4 / 600, jCardList.get(8), .001);
+ Assert.assertEquals(13. / 600, jCardList.get(15), .001);
+
+ Assert.assertTrue(new BigDecimal(jCardList.get(2)).round(mc).equals(new BigDecimal(64.0 / 6000).round(mc)));
+ Assert.assertTrue(new BigDecimal(jCardList.get(7)).round(mc).equals(new BigDecimal(7.0 / 7800).round(mc)));
+ Assert.assertTrue(new BigDecimal(jCardList.get(14)).round(mc).equals(new BigDecimal(112.0 / 7800).round(mc)));
+
+ }
+
+ private List<StatementPattern> getSpList(String query) throws MalformedQueryException {
+
+ SPARQLParser sp = new SPARQLParser();
+ ParsedQuery pq = sp.parseQuery(query, null);
+ TupleExpr te = pq.getTupleExpr();
+
+ return StatementPatternCollector.process(te);
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/CardinalityIdentityReducerTest.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/CardinalityIdentityReducerTest.java b/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/CardinalityIdentityReducerTest.java
new file mode 100644
index 0000000..eb480ad
--- /dev/null
+++ b/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/CardinalityIdentityReducerTest.java
@@ -0,0 +1,140 @@
+package mvm.rya.joinselect.mr;
+
+/*
+ * #%L
+ * mvm.rya.rya.prospector
+ * %%
+ * Copyright (C) 2014 - 2015 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.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import mvm.rya.joinselect.mr.utils.CardList;
+import mvm.rya.joinselect.mr.utils.TripleEntry;
+
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Value;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mrunit.mapreduce.ReduceDriver;
+import org.junit.Test;
+
+public class CardinalityIdentityReducerTest {
+
+ private static final String DELIM = "\u0000";
+
+ @Test
+ public void testCIReducerOneConstant() throws InterruptedException, IOException {
+
+ TripleEntry te = new TripleEntry(new Text("urn:gem:etype#1234"), new Text(""), new Text("subject"), new Text(""), new Text("object"));
+ CardList cL1 = new CardList(1, 2, 3, 0, 0, 0);
+ CardList cL2 = new CardList(4, 5, 6, 0, 0, 0);
+ CardList cl = new CardList(5, 7, 9, 0, 0, 0);
+ List<CardList> list = new ArrayList<CardList>();
+ list.add(cL1);
+ list.add(cL2);
+
+ Text row = new Text(te.getFirstPos().toString() + DELIM + te.getFirst().toString());
+ Mutation m1 = new Mutation(row);
+ m1.put(new Text(te.getKeyPos().toString() + "subject"), new Text(cl.getcardS().toString()), new Value(new byte[0]));
+ Mutation m2 = new Mutation(row);
+ m2.put(new Text(te.getKeyPos().toString() + "predicate"), new Text(cl.getcardP().toString()), new Value(new byte[0]));
+ Mutation m3 = new Mutation(row);
+ m3.put(new Text(te.getKeyPos().toString() + "object"), new Text(cl.getcardO().toString()), new Value(new byte[0]));
+ Text table = new Text("");
+
+ new ReduceDriver<TripleEntry,CardList,Text,Mutation>().withReducer(new JoinSelectStatisticsSum.CardinalityIdentityReducer()).withInput(te, list)
+ .withOutput(table, m1).withOutput(table, m2).withOutput(table, m3).runTest();
+
+ }
+
+ @Test
+ public void testCIReducerTwoConstant() throws InterruptedException, IOException {
+
+ TripleEntry te = new TripleEntry(new Text("urn:gem:etype#1234"), new Text("urn:gem#pred"), new Text("subject"), new Text("predicate"), new Text("object"));
+ CardList cL1 = new CardList(1, 2, 3, 0, 0, 0);
+ CardList cL2 = new CardList(4, 5, 6, 0, 0, 0);
+ CardList cl = new CardList(5, 7, 9, 0, 0, 0);
+ List<CardList> list = new ArrayList<CardList>();
+ list.add(cL1);
+ list.add(cL2);
+
+ Text row = new Text(te.getFirstPos().toString() + te.getSecondPos().toString() + DELIM + te.getFirst().toString() + DELIM + te.getSecond());
+ Mutation m1 = new Mutation(row);
+ m1.put(new Text(te.getKeyPos().toString() + "subject"), new Text(cl.getcardS().toString()), new Value(new byte[0]));
+ Mutation m2 = new Mutation(row);
+ m2.put(new Text(te.getKeyPos().toString() + "predicate"), new Text(cl.getcardP().toString()), new Value(new byte[0]));
+ Mutation m3 = new Mutation(row);
+ m3.put(new Text(te.getKeyPos().toString() + "object"), new Text(cl.getcardO().toString()), new Value(new byte[0]));
+ Text table = new Text("");
+
+ new ReduceDriver<TripleEntry,CardList,Text,Mutation>().withReducer(new JoinSelectStatisticsSum.CardinalityIdentityReducer()).withInput(te, list)
+ .withOutput(table, m1).withOutput(table, m2).withOutput(table, m3).runTest();
+
+ }
+
+ @Test
+ public void testJoinTwoVars() throws InterruptedException, IOException {
+
+ TripleEntry te = new TripleEntry(new Text("urn:gem:etype#1234"), new Text(""), new Text("subject"), new Text(""), new Text("predicateobject"));
+ CardList cL1 = new CardList(0, 0, 0, 1, 2, 3);
+ CardList cL2 = new CardList(0, 0, 0, 4, 5, 6);
+ CardList cl = new CardList(0, 0, 0, 5, 7, 9);
+ List<CardList> list = new ArrayList<CardList>();
+ list.add(cL1);
+ list.add(cL2);
+
+ Text row = new Text(te.getFirstPos().toString() + DELIM + te.getFirst().toString());
+ Mutation m1 = new Mutation(row);
+ m1.put(new Text(te.getKeyPos().toString() + "subjectpredicate"), new Text(cl.getcardSP().toString()), new Value(new byte[0]));
+ Mutation m2 = new Mutation(row);
+ m2.put(new Text(te.getKeyPos().toString() + "predicateobject"), new Text(cl.getcardPO().toString()), new Value(new byte[0]));
+ Mutation m3 = new Mutation(row);
+ m3.put(new Text(te.getKeyPos().toString() + "objectsubject"), new Text(cl.getcardSO().toString()), new Value(new byte[0]));
+ Text table = new Text("");
+
+ new ReduceDriver<TripleEntry,CardList,Text,Mutation>().withReducer(new JoinSelectStatisticsSum.CardinalityIdentityReducer()).withInput(te, list)
+ .withOutput(table, m1).withOutput(table, m2).withOutput(table, m3).runTest();
+
+ }
+
+ @Test
+ public void testJoinTwoVarsReverseOrder() throws InterruptedException, IOException {
+
+ TripleEntry te = new TripleEntry(new Text("urn:gem:etype#1234"), new Text(""), new Text("subject"), new Text(""), new Text("objectpredicate"));
+ CardList cL1 = new CardList(0, 0, 0, 1, 2, 3);
+ CardList cL2 = new CardList(0, 0, 0, 4, 5, 6);
+ CardList cl = new CardList(0, 0, 0, 5, 7, 9);
+ List<CardList> list = new ArrayList<CardList>();
+ list.add(cL1);
+ list.add(cL2);
+
+ Text row = new Text(te.getFirstPos().toString() + DELIM + te.getFirst().toString());
+ Mutation m1 = new Mutation(row);
+ m1.put(new Text("predicateobject" + "predicatesubject"), new Text(cl.getcardSP().toString()), new Value(new byte[0]));
+ Mutation m2 = new Mutation(row);
+ m2.put(new Text("predicateobject" + "objectpredicate"), new Text(cl.getcardPO().toString()), new Value(new byte[0]));
+ Mutation m3 = new Mutation(row);
+ m3.put(new Text("predicateobject" + "subjectobject"), new Text(cl.getcardSO().toString()), new Value(new byte[0]));
+ Text table = new Text("");
+
+ new ReduceDriver<TripleEntry,CardList,Text,Mutation>().withReducer(new JoinSelectStatisticsSum.CardinalityIdentityReducer()).withInput(te, list)
+ .withOutput(table, m1).withOutput(table, m2).withOutput(table, m3).runTest();
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/CardinalityMapperTest.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/CardinalityMapperTest.java b/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/CardinalityMapperTest.java
new file mode 100644
index 0000000..8647294
--- /dev/null
+++ b/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/CardinalityMapperTest.java
@@ -0,0 +1,75 @@
+package mvm.rya.joinselect.mr;
+
+/*
+ * #%L
+ * mvm.rya.rya.prospector
+ * %%
+ * Copyright (C) 2014 - 2015 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.io.IOException;
+
+import mvm.rya.joinselect.mr.JoinSelectProspectOutput;
+import mvm.rya.joinselect.mr.utils.CardinalityType;
+import mvm.rya.joinselect.mr.utils.CompositeType;
+import mvm.rya.joinselect.mr.utils.TripleCard;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Value;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mrunit.mapreduce.MapDriver;
+import org.junit.Test;
+
+public class CardinalityMapperTest {
+
+ private static final String DELIM = "\u0000";
+
+ public enum TripleValueType {
+ subject, predicate, object, subjectpredicate, predicateobject, subjectobject
+ }
+
+ @Test
+ public void testOutput() throws InterruptedException, IOException {
+
+ String s = "urn:gem:etype#1234";
+ String p = "urn:gem#pred";
+
+ Text t1 = new Text(TripleValueType.subject.name() + DELIM + s + DELIM + 1);
+ Text t2 = new Text(TripleValueType.predicate.name() + DELIM + p + DELIM + 2);
+ Text t3 = new Text(TripleValueType.subjectpredicate.name() + DELIM + s + DELIM + p + DELIM + 3);
+
+ byte[] b = new byte[0];
+ byte[] c = "25".getBytes();
+ byte[] d = "47".getBytes();
+ byte[] e = "15".getBytes();
+
+ Key key1 = new Key(t1.getBytes(), b, b, b, 1);
+ Key key2 = new Key(t2.getBytes(), b, b, b, 1);
+ Key key3 = new Key(t3.getBytes(), b, b, b, 1);
+ Value val1 = new Value(c);
+ Value val2 = new Value(d);
+ Value val3 = new Value(e);
+
+ // System.out.println("Keys are " + key1 + " and " + key2);
+
+ new MapDriver<Key,Value,CompositeType,TripleCard>().withMapper(new JoinSelectProspectOutput.CardinalityMapper()).withInput(key1, val1)
+ .withInput(key2, val2).withInput(key3, val3).withOutput(new CompositeType(s, 1), new TripleCard(new CardinalityType(25, "subject", 1)))
+ .withOutput(new CompositeType(p, 1), new TripleCard(new CardinalityType(47, "predicate", 2)))
+ .withOutput(new CompositeType(s + DELIM + p, 1), new TripleCard(new CardinalityType(15, "subjectpredicate", 3))).runTest();
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/FullTableSizeTest.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/FullTableSizeTest.java b/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/FullTableSizeTest.java
new file mode 100644
index 0000000..c070488
--- /dev/null
+++ b/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/FullTableSizeTest.java
@@ -0,0 +1,63 @@
+package mvm.rya.joinselect.mr;
+
+/*
+ * #%L
+ * mvm.rya.rya.prospector
+ * %%
+ * Copyright (C) 2014 - 2015 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.io.IOException;
+
+import mvm.rya.joinselect.mr.FullTableSize;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Mutation;
+import org.apache.accumulo.core.data.Value;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mrunit.mapreduce.MapReduceDriver;
+import org.junit.Test;
+
+//TODO fix table names!
+
+public class FullTableSizeTest {
+
+ private static final String DELIM = "\u0000";
+
+ @Test
+ public void testFullTableSize() throws IOException {
+
+ Value value = new Value(new byte[0]);
+
+ Mutation m = new Mutation(new Text("subjectpredicateobject" + DELIM + "FullTableCardinality"));
+ m.put(new Text("FullTableCardinality"), new Text("15"), new Value(new byte[0]));
+
+ new MapReduceDriver<Key, Value, Text, IntWritable, Text, Mutation>()
+ .withMapper(new FullTableSize.FullTableMapper()).withInput(new Key(new Text("entry1")), value)
+ .withInput(new Key(new Text("entry2")), value).withInput(new Key(new Text("entry3")), value)
+ .withInput(new Key(new Text("entry4")), value).withInput(new Key(new Text("entry5")), value)
+ .withInput(new Key(new Text("entry6")), value).withInput(new Key(new Text("entry7")), value)
+ .withInput(new Key(new Text("entry8")), value).withInput(new Key(new Text("entry9")), value)
+ .withInput(new Key(new Text("entry10")), value).withInput(new Key(new Text("entry11")), value)
+ .withInput(new Key(new Text("entry12")), value).withInput(new Key(new Text("entry13")), value)
+ .withInput(new Key(new Text("entry14")), value).withInput(new Key(new Text("entry15")), value)
+ .withCombiner(new FullTableSize.FullTableCombiner()).withReducer(new FullTableSize.FullTableReducer())
+ .withOutput(new Text(""), m).runTest();
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/JoinReducerTest.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/JoinReducerTest.java b/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/JoinReducerTest.java
new file mode 100644
index 0000000..5d2acb7
--- /dev/null
+++ b/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/JoinReducerTest.java
@@ -0,0 +1,123 @@
+package mvm.rya.joinselect.mr;
+
+/*
+ * #%L
+ * mvm.rya.rya.prospector
+ * %%
+ * Copyright (C) 2014 - 2015 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.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import mvm.rya.joinselect.mr.JoinSelectAggregate;
+import mvm.rya.joinselect.mr.utils.CardList;
+import mvm.rya.joinselect.mr.utils.CardinalityType;
+import mvm.rya.joinselect.mr.utils.CompositeType;
+import mvm.rya.joinselect.mr.utils.TripleCard;
+import mvm.rya.joinselect.mr.utils.TripleEntry;
+
+import org.apache.hadoop.mrunit.mapreduce.ReduceDriver;
+import org.junit.Test;
+
+public class JoinReducerTest {
+
+ private static final String DELIM = "\u0000";
+
+ @Test
+ public void testSingleConstCard() throws InterruptedException, IOException {
+
+ CompositeType ct = new CompositeType("urn:gem:etype#1234", 1);
+ TripleEntry te = new TripleEntry("urn:gem#pred", "urn:gem:etype#4567", "predicate", "object", "subject");
+ CardinalityType c5 = new CardinalityType(45, "object", 0);
+ CardinalityType c1 = new CardinalityType(25, "subject", 2);
+ CardinalityType c2 = new CardinalityType(27, "predicate", 2);
+ CardinalityType c3 = new CardinalityType(29, "object", 2);
+ CardinalityType c4 = new CardinalityType(31, "predicate", 1);
+ List<TripleCard> list = new ArrayList<TripleCard>();
+ list.add(new TripleCard(c1));
+ list.add(new TripleCard(c2));
+ list.add(new TripleCard(c3));
+ list.add(new TripleCard(c4));
+ list.add(new TripleCard(c5));
+ list.add(new TripleCard(te));
+ System.out.println("List is " + list);
+
+ new ReduceDriver<CompositeType,TripleCard,TripleEntry,CardList>().withReducer(new JoinSelectAggregate.JoinReducer()).withInput(ct, list)
+ .withOutput(te, new CardList(25, 31, 45, 0, 0, 0)).runTest();
+
+ }
+
+ @Test
+ public void testTwoTripleEntry() throws InterruptedException, IOException {
+
+ CompositeType ct = new CompositeType("urn:gem:etype#1234", 1);
+ TripleEntry te1 = new TripleEntry("urn:gem#pred", "urn:gem:etype#4567", "predicate", "object", "subject");
+ TripleEntry te2 = new TripleEntry("urn:gem#8910", "urn:gem:etype#4567", "subject", "predicate", "object");
+ CardinalityType c5 = new CardinalityType(45, "object", 0);
+ CardinalityType c1 = new CardinalityType(25, "subject", 2);
+ CardinalityType c2 = new CardinalityType(27, "predicate", 2);
+ CardinalityType c3 = new CardinalityType(29, "object", 2);
+ CardinalityType c4 = new CardinalityType(31, "predicate", 1);
+ List<TripleCard> list = new ArrayList<TripleCard>();
+ list.add(new TripleCard(c1));
+ list.add(new TripleCard(c2));
+ list.add(new TripleCard(c3));
+ list.add(new TripleCard(c4));
+ list.add(new TripleCard(c5));
+ list.add(new TripleCard(te1));
+ list.add(new TripleCard(te2));
+ System.out.println("List is " + list);
+
+ new ReduceDriver<CompositeType,TripleCard,TripleEntry,CardList>().withReducer(new JoinSelectAggregate.JoinReducer()).withInput(ct, list)
+ .withOutput(te1, new CardList(25, 31, 45, 0, 0, 0)).withOutput(te2, new CardList(25, 31, 45, 0, 0, 0)).runTest();
+
+ }
+
+ @Test
+ public void testTwoConstCard() throws InterruptedException, IOException {
+
+ CompositeType ct1 = new CompositeType("urn:gem#pred" + DELIM + "urn:gem:etype#1234", 1);
+ TripleEntry te1 = new TripleEntry("uri:testSubject", "", "subject", "", "predicateobject");
+ TripleEntry te2 = new TripleEntry("uri:testSubject", "", "subject", "", "objectpredicate");
+
+ CardinalityType c5 = new CardinalityType(45, "subjectobject", 0);
+ CardinalityType c1 = new CardinalityType(25, "subjectobject", 2);
+ CardinalityType c2 = new CardinalityType(27, "predicateobject", 5);
+ CardinalityType c3 = new CardinalityType(29, "predicateobject", 2);
+ CardinalityType c4 = new CardinalityType(31, "subjectpredicate", 1);
+ CardinalityType c6 = new CardinalityType(56, "subjectpredicate", 2);
+
+ List<TripleCard> list1 = new ArrayList<TripleCard>();
+
+ list1.add(new TripleCard(c1));
+ list1.add(new TripleCard(c2));
+ list1.add(new TripleCard(c3));
+ list1.add(new TripleCard(c4));
+ list1.add(new TripleCard(c5));
+ list1.add(new TripleCard(c6));
+ list1.add(new TripleCard(te1));
+ list1.add(new TripleCard(te2));
+
+ // System.out.println("List is " + list);
+
+ new ReduceDriver<CompositeType,TripleCard,TripleEntry,CardList>().withReducer(new JoinSelectAggregate.JoinReducer()).withInput(ct1, list1)
+ .withOutput(te1, new CardList(0, 0, 0, 31, 29, 45)).withOutput(te2, new CardList(0, 0, 0, 31, 29, 45)).runTest();
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/JoinSelectMapperTest.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/JoinSelectMapperTest.java b/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/JoinSelectMapperTest.java
new file mode 100644
index 0000000..e9281da
--- /dev/null
+++ b/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/JoinSelectMapperTest.java
@@ -0,0 +1,93 @@
+package mvm.rya.joinselect.mr;
+
+/*
+ * #%L
+ * mvm.rya.rya.prospector
+ * %%
+ * Copyright (C) 2014 - 2015 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.io.IOException;
+import java.util.Map;
+
+import mvm.rya.joinselect.mr.JoinSelectSpoTableOutput;
+import mvm.rya.joinselect.mr.utils.CompositeType;
+import mvm.rya.joinselect.mr.utils.TripleCard;
+import mvm.rya.joinselect.mr.utils.TripleEntry;
+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.triple.TripleRow;
+import mvm.rya.api.resolver.triple.TripleRowResolver;
+import mvm.rya.api.resolver.triple.TripleRowResolverException;
+import mvm.rya.api.resolver.triple.impl.WholeRowTripleResolver;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Value;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mrunit.mapreduce.MapDriver;
+import org.junit.Test;
+
+public class JoinSelectMapperTest {
+
+ private static final String DELIM = "\u0000";
+
+ @Test
+ public void testOutput() throws TripleRowResolverException, IOException {
+
+ RyaStatement rya = new RyaStatement(new RyaURI("urn:gem:etype#1234"), new RyaURI("urn:gem#pred"), new RyaType("mydata1"));
+ Text s = new Text(rya.getSubject().getData());
+ Text p = new Text(rya.getPredicate().getData());
+ Text o = new Text(rya.getObject().getData());
+ Text sp = new Text(rya.getSubject().getData() + DELIM + rya.getPredicate().getData());
+ Text so = new Text(rya.getSubject().getData() + DELIM + rya.getObject().getData());
+ Text po = new Text(rya.getPredicate().getData() + DELIM + rya.getObject().getData());
+ Text ps = new Text(rya.getPredicate().getData() + DELIM + rya.getSubject().getData());
+ Text op = new Text(rya.getObject().getData() + DELIM + rya.getPredicate().getData());
+ Text os = new Text(rya.getObject().getData() + DELIM + rya.getSubject().getData());
+
+ TripleEntry t1 = new TripleEntry(s, p, new Text("subject"), new Text("predicate"), new Text("object"));
+ TripleEntry t2 = new TripleEntry(p, o, new Text("predicate"), new Text("object"), new Text("subject"));
+ TripleEntry t3 = new TripleEntry(o, s, new Text("object"), new Text("subject"), new Text("predicate"));
+ TripleEntry t4 = new TripleEntry(o, new Text(""), new Text("object"), new Text(""), new Text("subjectpredicate"));
+ TripleEntry t5 = new TripleEntry(p, new Text(""), new Text("predicate"), new Text(""), new Text("objectsubject"));
+ TripleEntry t6 = new TripleEntry(s, new Text(""), new Text("subject"), new Text(""), new Text("predicateobject"));
+ TripleEntry t7 = new TripleEntry(s, new Text(""), new Text("subject"), new Text(""), new Text("objectpredicate"));
+ TripleEntry t8 = new TripleEntry(p, new Text(""), new Text("predicate"), new Text(""), new Text("subjectobject"));
+ TripleEntry t9 = new TripleEntry(o, new Text(""), new Text("object"), new Text(""), new Text("predicatesubject"));
+
+ TripleRowResolver trr = new WholeRowTripleResolver();
+ Map<TABLE_LAYOUT,TripleRow> map = trr.serialize(rya);
+ System.out.println(map);
+ TripleRow tr = map.get(TABLE_LAYOUT.SPO);
+ System.out.println("Triple row is" + tr);
+ System.out.println("ColumnV is " + tr.getTimestamp());
+ byte[] b = new byte[0];
+ Key key = new Key(tr.getRow(), tr.getColumnFamily(), tr.getColumnQualifier(), b, 1);
+ Value val = new Value(b);
+
+ new MapDriver<Key,Value,CompositeType,TripleCard>().withMapper(new JoinSelectSpoTableOutput.JoinSelectMapper()).withInput(key, val)
+ .withOutput(new CompositeType(o, new IntWritable(2)), new TripleCard(t1)).withOutput(new CompositeType(s, new IntWritable(2)), new TripleCard(t2))
+ .withOutput(new CompositeType(p, new IntWritable(2)), new TripleCard(t3)).withOutput(new CompositeType(po, new IntWritable(2)), new TripleCard(t6))
+ .withOutput(new CompositeType(so, new IntWritable(2)), new TripleCard(t5)).withOutput(new CompositeType(sp, new IntWritable(2)), new TripleCard(t4))
+ .withOutput(new CompositeType(op, new IntWritable(2)), new TripleCard(t7)).withOutput(new CompositeType(os, new IntWritable(2)), new TripleCard(t8))
+ .withOutput(new CompositeType(ps, new IntWritable(2)), new TripleCard(t9)).runTest();
+
+ }
+
+}
http://git-wip-us.apache.org/repos/asf/incubator-rya/blob/92ddfa59/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/JoinSelectProspectOutputTest.java
----------------------------------------------------------------------
diff --git a/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/JoinSelectProspectOutputTest.java b/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/JoinSelectProspectOutputTest.java
new file mode 100644
index 0000000..c00f8ff
--- /dev/null
+++ b/extras/rya.prospector/src/test/java/mvm/rya/joinselect/mr/JoinSelectProspectOutputTest.java
@@ -0,0 +1,69 @@
+package mvm.rya.joinselect.mr;
+
+import static org.junit.Assert.*;
+
+import org.junit.Test;
+
+import java.io.IOException;
+
+import mvm.rya.joinselect.mr.JoinSelectProspectOutput;
+import mvm.rya.joinselect.mr.utils.CardinalityType;
+import mvm.rya.joinselect.mr.utils.CompositeType;
+import mvm.rya.joinselect.mr.utils.TripleCard;
+
+import org.apache.accumulo.core.data.Key;
+import org.apache.accumulo.core.data.Value;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mrunit.mapreduce.MapDriver;
+import org.junit.Test;
+
+public class JoinSelectProspectOutputTest {
+
+ private static final String DELIM = "\u0000";
+
+ public enum TripleValueType {
+ subject, predicate, object, subjectpredicate, predicateobject, subjectobject
+ }
+
+ @Test
+ public void testOutput() throws InterruptedException, IOException {
+
+ String s = "urn:gem:etype#1234";
+ String p = "urn:gem#pred";
+
+ String ts = "798497748386999999";
+
+ Text t1 = new Text(TripleValueType.subject.name() + DELIM + s + DELIM + 1);
+ Text t2 = new Text(TripleValueType.predicate.name() + DELIM + p + DELIM + 2);
+ Text t3 = new Text(TripleValueType.subjectpredicate.name() + DELIM + s + DELIM + p + DELIM + ts);
+
+ byte[] b = new byte[0];
+ byte[] c = "25".getBytes();
+ byte[] d = "47".getBytes();
+ byte[] e = "15".getBytes();
+
+ Key key1 = new Key(t1.getBytes(), b, b, b, 1);
+ Key key2 = new Key(t2.getBytes(), b, b, b, 1);
+ Key key3 = new Key(t3.getBytes(), b, b, b, 1);
+ Value val1 = new Value(c);
+ Value val2 = new Value(d);
+ Value val3 = new Value(e);
+
+
+
+ // System.out.println("Keys are " + key1 + " and " + key2);
+
+ new MapDriver<Key, Value, CompositeType, TripleCard>()
+ .withMapper(new JoinSelectProspectOutput.CardinalityMapper())
+ .withInput(key1, val1)
+ .withInput(key2, val2)
+ .withInput(key3, val3)
+ .withOutput(new CompositeType(s, 1), new TripleCard(new CardinalityType(25, "subject", 1)))
+ .withOutput(new CompositeType(p, 1), new TripleCard(new CardinalityType(47, "predicate", 2)))
+ .withOutput(new CompositeType(s + DELIM + p, 1),
+ new TripleCard(new CardinalityType(15, "subjectpredicate", Long.parseLong(ts)))).runTest();
+
+ }
+
+}