Menu
▾
▴
[Bigdata-commit] SF.net SVN: bigdata:[4358] branches/QUADS_QUERY_BRANCH/bigdata/src
|
From: <tho...@us...> - 2011-03-31 17:53:47
|
Revision: 4358
http://bigdata.svn.sourceforge.net/bigdata/?rev=4358&view=rev
Author: thompsonbry
Date: 2011-03-31 17:53:40 +0000 (Thu, 31 Mar 2011)
Log Message:
-----------
Added support for SELECT and CONSTRAINTS to SubqueryOp.
Added unit tests for simple subquery join, subquery join with SELECT, and subquery join with CONSTRAINTS.
Modified the method signature for BOpUtility#copy(...)
Modified Paths:
--------------
branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/BOpUtility.java
branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/bset/CopyOp.java
branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/controller/AbstractSubqueryOp.java
branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/controller/SubqueryOp.java
branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/engine/StandaloneChainedRunningQuery.java
branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/joinGraph/rto/JoinGraph.java
branches/QUADS_QUERY_BRANCH/bigdata/src/test/com/bigdata/bop/controller/TestSubqueryOp.java
Modified: branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/BOpUtility.java
===================================================================
--- branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/BOpUtility.java 2011-03-31 17:11:26 UTC (rev 4357)
+++ branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/BOpUtility.java 2011-03-31 17:53:40 UTC (rev 4358)
@@ -888,29 +888,35 @@
}
- /**
- * Copy binding sets from the source to the sink(s).
- *
- * @param source
- * The source.
- * @param sink
- * The sink (required).
- * @param sink2
- * Another sink (optional).
- * @param constraints
- * Binding sets which fail these constraints will NOT be copied
- * (optional).
- * @param stats
- * The {@link BOpStats#chunksIn} and {@link BOpStats#unitsIn}
- * will be updated during the copy (optional).
- *
- * @return The #of binding sets copied.
- */
+ /**
+ * Copy binding sets from the source to the sink(s).
+ *
+ * @param source
+ * The source.
+ * @param sink
+ * The sink (required).
+ * @param sink2
+ * Another sink (optional).
+ * @param select
+ * The variables to be retained (optional). When not specified,
+ * all variables will be retained.
+ * @param constraints
+ * Binding sets which fail these constraints will NOT be copied
+ * (optional).
+ * @param stats
+ * The {@link BOpStats#chunksIn} and {@link BOpStats#unitsIn}
+ * will be updated during the copy (optional).
+ *
+ * @return The #of binding sets copied.
+ */
static public long copy(
final IAsynchronousIterator<IBindingSet[]> source,
final IBlockingBuffer<IBindingSet[]> sink,
final IBlockingBuffer<IBindingSet[]> sink2,
- final IConstraint[] constraints, final BOpStats stats) {
+ final IVariable<?>[] select,//
+ final IConstraint[] constraints, //
+ final BOpStats stats//
+ ) {
long nout = 0;
@@ -926,9 +932,10 @@
}
- // apply optional constraints.
- final IBindingSet[] tmp = applyConstraints(chunk,constraints);
-
+ // apply optional constraints and optional SELECT.
+ final IBindingSet[] tmp = applyConstraints(chunk, select,
+ constraints);
+
// System.err.println("Copying: "+Arrays.toString(tmp));
// copy accepted binding sets to the default sink.
@@ -955,18 +962,23 @@
*
* @param chunk
* A chunk of binding sets.
+ * @param select
+ * The variables to be retained (optional). When not specified,
+ * all variables will be retained.
* @param constraints
* The constraints (optional).
*
* @return The dense chunk of binding sets.
*/
static private IBindingSet[] applyConstraints(final IBindingSet[] chunk,
+ final IVariable<?>[] select,
final IConstraint[] constraints) {
- if (constraints == null) {
+ if (constraints == null && select == null) {
/*
- * No constraints, copy all binding sets.
+ * No constraints and everything is selected, so just return the
+ * caller's chunk.
*/
return chunk;
@@ -983,14 +995,23 @@
for (int i = 0; i < chunk.length; i++) {
- final IBindingSet bindingSet = chunk[i];
+ IBindingSet bindingSet = chunk[i];
- if (BOpUtility.isConsistent(constraints, bindingSet)) {
+ if (constraints != null
+ && !BOpUtility.isConsistent(constraints, bindingSet)) {
- t[j++] = bindingSet;
+ continue;
}
+ if (select != null) {
+
+ bindingSet = bindingSet.copy(select);
+
+ }
+
+ t[j++] = bindingSet;
+
}
if (j != chunk.length) {
Modified: branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/bset/CopyOp.java
===================================================================
--- branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/bset/CopyOp.java 2011-03-31 17:11:26 UTC (rev 4357)
+++ branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/bset/CopyOp.java 2011-03-31 17:53:40 UTC (rev 4358)
@@ -36,6 +36,7 @@
import com.bigdata.bop.BOpUtility;
import com.bigdata.bop.IBindingSet;
import com.bigdata.bop.IConstraint;
+import com.bigdata.bop.IVariable;
import com.bigdata.bop.PipelineOp;
import com.bigdata.bop.engine.BOpStats;
import com.bigdata.bop.engine.IChunkAccessor;
@@ -65,16 +66,22 @@
public interface Annotations extends PipelineOp.Annotations {
/**
+ * An optional {@link IVariable}[] which specifies which variables will
+ * have their bindings copied.
+ */
+ String SELECT = CopyOp.class.getName() + ".select";
+
+ /**
* An optional {@link IConstraint}[] which places restrictions on the
* legal patterns in the variable bindings.
*/
- String CONSTRAINTS = (CopyOp.class.getName() + ".constraints").intern();
+ String CONSTRAINTS = CopyOp.class.getName() + ".constraints";
/**
* An optional {@link IBindingSet}[] to be used <strong>instead</strong>
* of the default source.
*/
- String BINDING_SETS = (CopyOp.class.getName() + ".bindingSets").intern();
+ String BINDING_SETS = CopyOp.class.getName() + ".bindingSets";
}
@@ -98,6 +105,15 @@
}
/**
+ * @see Annotations#SELECT
+ */
+ public IVariable<?>[] getSelect() {
+
+ return getProperty(Annotations.SELECT, null/* defaultValue */);
+
+ }
+
+ /**
* @see Annotations#CONSTRAINTS
*/
public IConstraint[] constraints() {
@@ -156,6 +172,8 @@
final BOpStats stats = context.getStats();
+ final IVariable<?>[] select = op.getSelect();
+
final IConstraint[] constraints = op.constraints();
try {
@@ -168,12 +186,13 @@
BOpUtility.copy(
new ThickAsynchronousIterator<IBindingSet[]>(
new IBindingSet[][] { bindingSets }), sink,
- sink2, constraints, stats);
+ sink2, select, constraints, stats);
} else {
// copy binding sets from the source.
- BOpUtility.copy(source, sink, sink2, constraints, stats);
+ BOpUtility.copy(source, sink, sink2, select, constraints,
+ stats);
}
Modified: branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/controller/AbstractSubqueryOp.java
===================================================================
--- branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/controller/AbstractSubqueryOp.java 2011-03-31 17:11:26 UTC (rev 4357)
+++ branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/controller/AbstractSubqueryOp.java 2011-03-31 17:53:40 UTC (rev 4358)
@@ -45,7 +45,6 @@
import com.bigdata.bop.engine.IRunningQuery;
import com.bigdata.bop.engine.QueryEngine;
import com.bigdata.relation.accesspath.IAsynchronousIterator;
-import com.bigdata.service.IBigdataFederation;
import com.bigdata.util.concurrent.LatchedExecutor;
/**
@@ -367,8 +366,8 @@
// Copy solutions from the subquery to the query.
BOpUtility.copy(subquerySolutionItr, parentContext
- .getSink(), null/* sink2 */, null/* constraints */,
- null/* stats */);
+ .getSink(), null/* sink2 */, null/* select */,
+ null/* constraints */, null/* stats */);
// wait for the subquery.
runningSubquery.get();
Modified: branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/controller/SubqueryOp.java
===================================================================
--- branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/controller/SubqueryOp.java 2011-03-31 17:11:26 UTC (rev 4357)
+++ branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/controller/SubqueryOp.java 2011-03-31 17:53:40 UTC (rev 4358)
@@ -35,11 +35,12 @@
import com.bigdata.bop.BOpContext;
import com.bigdata.bop.BOpUtility;
import com.bigdata.bop.IBindingSet;
+import com.bigdata.bop.IConstraint;
+import com.bigdata.bop.IVariable;
import com.bigdata.bop.NV;
import com.bigdata.bop.PipelineOp;
import com.bigdata.bop.engine.IRunningQuery;
import com.bigdata.bop.engine.QueryEngine;
-import com.bigdata.bop.join.JoinAnnotations;
import com.bigdata.relation.accesspath.IAsynchronousIterator;
/**
@@ -54,18 +55,13 @@
*
* @todo Parallel evaluation of subqueries is not implemented. What is the
* appropriate parallelism for this operator? More parallelism should
- * reduce latency but could increase the memory burden. Review this
- * decision once we have the RWStore operating as a binding set buffer on
- * the Java process heap.
+ * reduce latency but could increase the memory burden.
*
- * @todo Rename as SubqueryPipelineJoinOp and support for SELECT and CONSTRAINTS.
+ * @todo Rename as SubqueryPipelineJoinOp.
*
* @author <a href="mailto:tho...@us...">Bryan Thompson</a>
*
* @see AbstractSubqueryOp
- *
- * FIXME Support {@link JoinAnnotations#SELECT}
- * FIXME Support {@link JoinAnnotations#CONSTRAINTS}
*/
public class SubqueryOp extends PipelineOp {
@@ -168,16 +164,23 @@
*/
private static class ControllerTask implements Callable<Void> {
-// private final SubqueryOp controllerOp;
private final BOpContext<IBindingSet> context;
// private final List<FutureTask<IRunningQuery>> tasks = new LinkedList<FutureTask<IRunningQuery>>();
// private final CountDownLatch latch;
private final boolean optional;
// private final int nparallel;
+// /** The {@link SubqueryOp}. */
+// private final SubqueryOp controllerOp;
+ /** The subquery which is evaluated for each input binding set. */
private final PipelineOp subquery;
+ /** The selected variables for the output binding sets (optional). */
+ private final IVariable<?>[] selectVars;
+ /** The optional constraints on the join. */
+ private final IConstraint[] constraints;
// private final Executor executor;
- public ControllerTask(final SubqueryOp controllerOp, final BOpContext<IBindingSet> context) {
+ public ControllerTask(final SubqueryOp controllerOp,
+ final BOpContext<IBindingSet> context) {
if (controllerOp == null)
throw new IllegalArgumentException();
@@ -199,6 +202,12 @@
this.subquery = (PipelineOp) controllerOp
.getRequiredProperty(Annotations.SUBQUERY);
+ this.selectVars = (IVariable<?>[]) controllerOp
+ .getProperty(Annotations.SELECT);
+
+ this.constraints = (IConstraint[]) controllerOp
+ .getProperty(Annotations.CONSTRAINTS);
+
// this.executor = new LatchedExecutor(context.getIndexManager()
// .getExecutorService(), nparallel);
@@ -379,25 +388,6 @@
final QueryEngine queryEngine = parentContext.getRunningQuery()
.getQueryEngine();
-
-// final BOp startOp = BOpUtility.getPipelineStart(subQueryOp);
-//
-// final int startId = startOp.getId();
-//
-// final UUID queryId = UUID.randomUUID();
-//
-// // execute the subquery, passing in the source binding set.
-// runningSubquery = queryEngine
-// .eval(
-// queryId,
-// (PipelineOp) subQueryOp,
-// new LocalChunkMessage<IBindingSet>(
-// queryEngine,
-// queryId,
-// startId,
-// -1 /* partitionId */,
-// new ThickAsynchronousIterator<IBindingSet[]>(
-// new IBindingSet[][] { new IBindingSet[] { bset } })));
runningSubquery = queryEngine.eval((PipelineOp) subQueryOp,
bset);
@@ -409,9 +399,10 @@
subquerySolutionItr = runningSubquery.iterator();
// Copy solutions from the subquery to the query.
- ncopied = BOpUtility.copy(subquerySolutionItr,
- parentContext.getSink(), null/* sink2 */,
- null/* constraints */, null/* stats */);
+ ncopied = BOpUtility.copy(subquerySolutionItr,
+ parentContext.getSink(), null/* sink2 */,
+ selectVars, constraints, parentContext
+ .getStats());
// wait for the subquery to halt / test for errors.
runningSubquery.get();
Modified: branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/engine/StandaloneChainedRunningQuery.java
===================================================================
--- branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/engine/StandaloneChainedRunningQuery.java 2011-03-31 17:11:26 UTC (rev 4357)
+++ branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/engine/StandaloneChainedRunningQuery.java 2011-03-31 17:53:40 UTC (rev 4358)
@@ -232,7 +232,8 @@
try {
BOpUtility.copy(msg.getChunkAccessor().iterator(), sink,
- null/* sink2 */, null/* constraints */, //
+ null/* sink2 */, null/* selectVars */,
+ null/* constraints */, //
null/* stats */
);
Modified: branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/joinGraph/rto/JoinGraph.java
===================================================================
--- branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/joinGraph/rto/JoinGraph.java 2011-03-31 17:11:26 UTC (rev 4357)
+++ branches/QUADS_QUERY_BRANCH/bigdata/src/java/com/bigdata/bop/joinGraph/rto/JoinGraph.java 2011-03-31 17:53:40 UTC (rev 4358)
@@ -362,9 +362,10 @@
subquerySolutionItr = runningQuery.iterator();
// Copy solutions from the subquery to the query.
- final long nout = BOpUtility.copy(subquerySolutionItr,
- parentContext.getSink(), null/* sink2 */,
- null/* constraints */, null/* stats */);
+ final long nout = BOpUtility
+ .copy(subquerySolutionItr, parentContext.getSink(),
+ null/* sink2 */, null/* selectVars */,
+ null/* constraints */, null/* stats */);
// System.out.println("nout=" + nout);
Modified: branches/QUADS_QUERY_BRANCH/bigdata/src/test/com/bigdata/bop/controller/TestSubqueryOp.java
===================================================================
--- branches/QUADS_QUERY_BRANCH/bigdata/src/test/com/bigdata/bop/controller/TestSubqueryOp.java 2011-03-31 17:11:26 UTC (rev 4357)
+++ branches/QUADS_QUERY_BRANCH/bigdata/src/test/com/bigdata/bop/controller/TestSubqueryOp.java 2011-03-31 17:53:40 UTC (rev 4358)
@@ -27,6 +27,8 @@
package com.bigdata.bop.controller;
+import java.util.LinkedList;
+import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;
@@ -51,6 +53,7 @@
import com.bigdata.bop.bset.ConditionalRoutingOp;
import com.bigdata.bop.bset.StartOp;
import com.bigdata.bop.constraint.Constraint;
+import com.bigdata.bop.constraint.EQConstant;
import com.bigdata.bop.constraint.NEConstant;
import com.bigdata.bop.engine.BOpStats;
import com.bigdata.bop.engine.IChunkMessage;
@@ -164,6 +167,323 @@
}
/**
+ * Unit test for a simple join.
+ */
+ public void test_join() throws Exception {
+
+// final int startId = 1;
+ final int joinId = 2;
+ final int predId = 3;
+ final int subqueryId = 4;
+
+ final IVariable<?> x = Var.var("x");
+ final IVariable<?> y = Var.var("y");
+
+ final Predicate<E> predOp = new Predicate<E>(//
+ new IVariableOrConstant[] { //
+ new Constant<String>("John"), x }, //
+ NV.asMap(new NV[] {//
+ new NV(Predicate.Annotations.RELATION_NAME,
+ new String[] { namespace }),//
+ new NV(Predicate.Annotations.BOP_ID, predId),//
+ new NV(Annotations.TIMESTAMP,
+ ITx.READ_COMMITTED),//
+ }));
+
+ // the subquery (basically, an access path read with "x" unbound).
+ final PipelineJoin<E> subquery = new PipelineJoin<E>(
+ new BOp[] { },//
+ new NV(Predicate.Annotations.BOP_ID, joinId),//
+ new NV(PipelineJoin.Annotations.PREDICATE, predOp));
+
+ // the hash-join against the subquery.
+ final SubqueryOp subqueryOp = new SubqueryOp(
+ new BOp[] {},//
+ new NV(Predicate.Annotations.BOP_ID, subqueryId),//
+ new NV(SubqueryOp.Annotations.SUBQUERY, subquery)//
+ );
+
+ final PipelineOp query = subqueryOp;
+
+ // the expected solutions.
+ final IBindingSet[] expected = new IBindingSet[] {//
+ new ArrayBindingSet(//
+ new IVariable[] { x },//
+ new IConstant[] { new Constant<String>("Mary") }//
+ ),//
+ new ArrayBindingSet(//
+ new IVariable[] { x, y },//
+ new IConstant[] { new Constant<String>("Brad"),
+ new Constant<String>("Fred"),
+ }//
+ ),//
+ };
+
+ /*
+ * Setup the input binding sets. Each input binding set MUST provide
+ * binding for the join variable(s).
+ */
+ final IBindingSet[] initialBindingSets;
+ {
+ final List<IBindingSet> list = new LinkedList<IBindingSet>();
+
+ IBindingSet tmp;
+
+ tmp = new HashBindingSet();
+ tmp.set(x, new Constant<String>("Brad"));
+ tmp.set(y, new Constant<String>("Fred"));
+ list.add(tmp);
+
+ tmp = new HashBindingSet();
+ tmp.set(x, new Constant<String>("Mary"));
+ list.add(tmp);
+
+ initialBindingSets = list.toArray(new IBindingSet[0]);
+
+ }
+
+ final IRunningQuery runningQuery = queryEngine.eval(query,
+ initialBindingSets);
+
+ TestQueryEngine.assertSameSolutionsAnyOrder(expected, runningQuery);
+
+ {
+ final BOpStats stats = runningQuery.getStats().get(
+ subqueryId);
+ assertEquals(2L, stats.chunksIn.get());
+ assertEquals(2L, stats.unitsIn.get());
+ assertEquals(2L, stats.unitsOut.get());
+ assertEquals(2L, stats.chunksOut.get());
+ }
+ {
+ // // access path
+ // assertEquals(0L, stats.accessPathDups.get());
+ // assertEquals(1L, stats.accessPathCount.get());
+ // assertEquals(1L, stats.accessPathChunksIn.get());
+ // assertEquals(2L, stats.accessPathUnitsIn.get());
+ }
+
+ assertTrue(runningQuery.isDone());
+ assertFalse(runningQuery.isCancelled());
+ runningQuery.get(); // verify nothing thrown.
+
+ }
+
+ /**
+ * Unit test for simple join with a constraint.
+ */
+ public void test_joinWithConstraint() throws Exception {
+
+// final int startId = 1;
+ final int joinId = 2;
+ final int predId = 3;
+ final int subqueryId = 4;
+
+ final IVariable<?> x = Var.var("x");
+ final IVariable<?> y = Var.var("y");
+
+// final IConstant<String>[] set = new IConstant[] {//
+// new Constant<String>("Fred"),//
+// };
+
+ final Predicate<E> predOp = new Predicate<E>(//
+ new IVariableOrConstant[] { //
+ new Constant<String>("John"), x }, //
+ NV.asMap(new NV[] {//
+ new NV(Predicate.Annotations.RELATION_NAME,
+ new String[] { namespace }),//
+ new NV(Predicate.Annotations.BOP_ID, predId),//
+ new NV(Annotations.TIMESTAMP,
+ ITx.READ_COMMITTED),//
+ }));
+
+ // the subquery (basically, an access path read with "x" unbound).
+ final PipelineJoin<E> subquery = new PipelineJoin<E>(
+ new BOp[] { },//
+ new NV(Predicate.Annotations.BOP_ID, joinId),//
+ new NV(PipelineJoin.Annotations.PREDICATE, predOp));
+
+ // the hash-join against the subquery.
+ final SubqueryOp subqueryOp = new SubqueryOp(
+ new BOp[] {},//
+ new NV(Predicate.Annotations.BOP_ID, subqueryId),//
+ new NV(SubqueryOp.Annotations.SUBQUERY, subquery),//
+ new NV(SubqueryOp.Annotations.CONSTRAINTS,
+ new IConstraint[] { Constraint
+ .wrap(new EQConstant(x, new Constant<String>("Brad"))),//
+ }));
+
+ final PipelineOp query = subqueryOp;
+
+ // the expected solutions.
+ final IBindingSet[] expected = new IBindingSet[] {//
+// new ArrayBindingSet(//
+// new IVariable[] { x },//
+// new IConstant[] { new Constant<String>("Mary") }//
+// ),//
+ new ArrayBindingSet(//
+ new IVariable[] { x, y },//
+ new IConstant[] { new Constant<String>("Brad"),
+ new Constant<String>("Fred"),
+ }//
+ ),//
+ };
+
+ /*
+ * Setup the input binding sets. Each input binding set MUST provide
+ * binding for the join variable(s).
+ */
+ final IBindingSet[] initialBindingSets;
+ {
+ final List<IBindingSet> list = new LinkedList<IBindingSet>();
+
+ IBindingSet tmp;
+
+ tmp = new HashBindingSet();
+ tmp.set(x, new Constant<String>("Brad"));
+ tmp.set(y, new Constant<String>("Fred"));
+ list.add(tmp);
+
+ tmp = new HashBindingSet();
+ tmp.set(x, new Constant<String>("Mary"));
+ list.add(tmp);
+
+ initialBindingSets = list.toArray(new IBindingSet[0]);
+
+ }
+
+ final IRunningQuery runningQuery = queryEngine.eval(query,
+ initialBindingSets);
+
+ TestQueryEngine.assertSameSolutionsAnyOrder(expected, runningQuery);
+
+ {
+ final BOpStats stats = runningQuery.getStats().get(
+ subqueryId);
+ assertEquals(2L, stats.chunksIn.get());
+ assertEquals(2L, stats.unitsIn.get());
+ assertEquals(1L, stats.unitsOut.get());
+ assertEquals(2L, stats.chunksOut.get());
+ }
+ {
+ // // access path
+ // assertEquals(0L, stats.accessPathDups.get());
+ // assertEquals(1L, stats.accessPathCount.get());
+ // assertEquals(1L, stats.accessPathChunksIn.get());
+ // assertEquals(2L, stats.accessPathUnitsIn.get());
+ }
+
+ assertTrue(runningQuery.isDone());
+ assertFalse(runningQuery.isCancelled());
+ runningQuery.get(); // verify nothing thrown.
+
+ }
+
+ /**
+ * Unit test for a simple join.
+ */
+ public void test_join_selectOnly_x() throws Exception {
+
+// final int startId = 1;
+ final int joinId = 2;
+ final int predId = 3;
+ final int subqueryId = 4;
+
+ final IVariable<?> x = Var.var("x");
+ final IVariable<?> y = Var.var("y");
+
+ final Predicate<E> predOp = new Predicate<E>(//
+ new IVariableOrConstant[] { //
+ new Constant<String>("John"), x }, //
+ NV.asMap(new NV[] {//
+ new NV(Predicate.Annotations.RELATION_NAME,
+ new String[] { namespace }),//
+ new NV(Predicate.Annotations.BOP_ID, predId),//
+ new NV(Annotations.TIMESTAMP,
+ ITx.READ_COMMITTED),//
+ }));
+
+ // the subquery (basically, an access path read with "x" unbound).
+ final PipelineJoin<E> subquery = new PipelineJoin<E>(
+ new BOp[] { },//
+ new NV(Predicate.Annotations.BOP_ID, joinId),//
+ new NV(PipelineJoin.Annotations.PREDICATE, predOp));
+
+ // the hash-join against the subquery.
+ final SubqueryOp subqueryOp = new SubqueryOp(
+ new BOp[] {},//
+ new NV(Predicate.Annotations.BOP_ID, subqueryId),//
+ new NV(SubqueryOp.Annotations.SELECT, new IVariable[]{x}),//
+ new NV(SubqueryOp.Annotations.SUBQUERY, subquery)//
+ );
+
+ final PipelineOp query = subqueryOp;
+
+ // the expected solutions.
+ final IBindingSet[] expected = new IBindingSet[] {//
+ new ArrayBindingSet(//
+ new IVariable[] { x },//
+ new IConstant[] { new Constant<String>("Mary") }//
+ ),//
+ new ArrayBindingSet(//
+ new IVariable[] { x },//
+ new IConstant[] { new Constant<String>("Brad"),
+// new Constant<String>("Fred"),
+ }//
+ ),//
+ };
+
+ /*
+ * Setup the input binding sets. Each input binding set MUST provide
+ * binding for the join variable(s).
+ */
+ final IBindingSet[] initialBindingSets;
+ {
+ final List<IBindingSet> list = new LinkedList<IBindingSet>();
+
+ IBindingSet tmp;
+
+ tmp = new HashBindingSet();
+ tmp.set(x, new Constant<String>("Brad"));
+ tmp.set(y, new Constant<String>("Fred"));
+ list.add(tmp);
+
+ tmp = new HashBindingSet();
+ tmp.set(x, new Constant<String>("Mary"));
+ list.add(tmp);
+
+ initialBindingSets = list.toArray(new IBindingSet[0]);
+
+ }
+
+ final IRunningQuery runningQuery = queryEngine.eval(query,
+ initialBindingSets);
+
+ TestQueryEngine.assertSameSolutionsAnyOrder(expected, runningQuery);
+
+ {
+ final BOpStats stats = runningQuery.getStats().get(
+ subqueryId);
+ assertEquals(2L, stats.chunksIn.get());
+ assertEquals(2L, stats.unitsIn.get());
+ assertEquals(2L, stats.unitsOut.get());
+ assertEquals(2L, stats.chunksOut.get());
+ }
+ {
+ // // access path
+ // assertEquals(0L, stats.accessPathDups.get());
+ // assertEquals(1L, stats.accessPathCount.get());
+ // assertEquals(1L, stats.accessPathChunksIn.get());
+ // assertEquals(2L, stats.accessPathUnitsIn.get());
+ }
+
+ assertTrue(runningQuery.isDone());
+ assertFalse(runningQuery.isCancelled());
+ runningQuery.get(); // verify nothing thrown.
+
+ }
+
+ /**
* Unit test for optional join group. Three joins are used and target a
* {@link SliceOp}. The 2nd and 3rd joins are embedded in an
* {@link SubqueryOp}.
This was sent by the SourceForge.net collaborative development platform, the world's largest Open Source development site.
|