com.bigdata.relation.rule.eval.pipeline
Class DistributedJoinMasterTask

java.lang.Object
  com.bigdata.relation.rule.eval.pipeline.JoinMasterTask
      com.bigdata.relation.rule.eval.pipeline.DistributedJoinMasterTask

All Implemented Interfaces:

IStepTask, IJoinMaster, Serializable, Remote, Callable<RuleStats>

public class DistributedJoinMasterTask
extends JoinMasterTask
implements Serializable

Implementation for distributed join execution.

Note: For query, this object MUST be executed locally on the client. This ensures that all data flows back to the client directly. For mutation, it is possible to submit this object to any service in the federation and each DistributedJoinTask will write directly on the scale-out view of the target IMutableRelation.

Version:

$Id: DistributedJoinMasterTask.java 3730 2010-10-05 19:57:25Z thompsonbry $

Author:

Bryan Thompson

See Also:

Serialized Form

Field Summary

Fields inherited from class com.bigdata.relation.rule.eval.pipeline.JoinMasterTask

DEBUG, joinNexus, joinNexusFactory, joinStats, log, masterUUID, order, rule, ruleState, ruleStats, solutionBuffer, tailCount

Constructor Summary

DistributedJoinMasterTask(IRule rule, IJoinNexus joinNexus, IBuffer<ISolution[]> buffer)

Method Summary

protected List<Future<Void>>	awaitFactoryFutures(List<Future> factoryTaskFutures) Await the JoinTaskFactoryTask Futures.
IBuffer<ISolution[]>	getSolutionBuffer() Returns the buffer specified to the ctor (overridden for distributed joins).
protected List<Future>	mapBindingSet(IBindingSet bindingSet) Map the given IBindingSet over the JoinTask(s) for the index partition(s) the span the IAccessPath for that IBindingSet in parallel.
protected List<Future<Void>>	start() Create and run the JoinTask(s) that will evaluate the first join dimension.

Methods inherited from class com.bigdata.relation.rule.eval.pipeline.JoinMasterTask

awaitAll, call, combineJoinStats, getUUID, newBindingSetIterator, report

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Detail

DistributedJoinMasterTask

public DistributedJoinMasterTask(IRule rule,
                                 IJoinNexus joinNexus,
                                 IBuffer<ISolution[]> buffer)

Parameters:

rule -

joinNexus -

buffer - The buffer on which the last DistributedJoinTask will write query ISolutions. However, it is ignored for mutation operations as each DistributedJoinTask for the last join dimension (there can be more than one if the index partition has more than one partition) will obtain and write on its own solution buffer in order to avoid moving all data through the master.

Throws:

UnsupportedOperationException - unless IJoinNexus.getIndexManager() reports an AbstractScaleOutFederation.

Method Detail

getSolutionBuffer

public IBuffer<ISolution[]> getSolutionBuffer()
                                       throws IOException

Description copied from class: JoinMasterTask

Returns the buffer specified to the ctor (overridden for distributed joins).

Specified by:

getSolutionBuffer in interface IJoinMaster

Overrides:

getSolutionBuffer in class JoinMasterTask

Throws:

IOException

start

protected final List<Future<Void>> start()
                                  throws Exception

Create and run the JoinTask(s) that will evaluate the first join dimension.

A JoinTask is created on the DataService for each index partition that is spanned by the IAccessPath for the first IPredicate in the evaluation order. Those JoinTask are run in parallel, so the actual parallelism for the first IPredicate is the #of index partitions spanned by its IAccessPath.

Returns:

The Future for each DistributedJoinTask created for the first join dimension (one per index partitions spanned by the predicate that is first in the evaluation order given the initial bindingSet for the rule).

Throws:

Exception

mapBindingSet

protected List<Future> mapBindingSet(IBindingSet bindingSet)
                              throws Exception

Map the given IBindingSet over the JoinTask(s) for the index partition(s) the span the IAccessPath for that IBindingSet in parallel.

Parameters:

bindingSet - The binding set.

Returns:

A list of Futures for the JoinTaskFactoryTask that will create the DistributedJoinTasks for the first join dimension.

Throws:

Exception - FIXME If a predicate defines an IAccessPathExpander then we DO NOT map the predicate. Instead, we use IJoinNexus#getTailAccessPath(IPredicate) and evaluate the IAccessPath with the layered IAccessPathExpander in process. If the IAccessPathExpander touches the index, it will be using an IClientIndex. While the IClientIndex is not nearly as efficient as using a local index partition, it will provide a view of the total key-range partitioned index.

do this for each join dimension for which an IAccessPathExpander is defined, including not only the first N join dimensions (handles free text search) but also an intermediate join dimension (requires that all source join tasks target a join task having a view of the scale-out index rather than mapping the task across the index partitions). FIXME The initial binding set should not be mapped across the index partitions for the first join dimension if IQueryOptions.isStable() is true (any parallel evaluation violates the stable constraint).

awaitFactoryFutures

protected List<Future<Void>> awaitFactoryFutures(List<Future> factoryTaskFutures)
                                          throws InterruptedException,
                                                 ExecutionExceptions

Await the JoinTaskFactoryTask Futures.

Note: the result for a JoinTaskFactoryTask Future is a DistributedJoinTask Future.

Parameters:

factoryTaskFutures - A list of Futures, with one Future for each index partition that is spanned by the IAccessPath for the first IPredicate in the evaluation order.

Returns:

A list of DistributedJoinTask Futures. There will be one element in the list for each JoinTaskFactoryTask Future in the caller's list. The elements will be in the same order.

Throws:

InterruptedException - if the master itself was interrupted.

ExecutionExceptions - if any of the factory tasks fail.