com.bigdata.bop.engine

Class ChunkedRunningQuery

java.lang.Object

com.bigdata.bop.engine.AbstractRunningQuery

com.bigdata.bop.engine.ChunkedRunningQuery

All Implemented Interfaces:

IRunningQuery, IQueryContext, IHaltable<Void>, Future<Void>

Direct Known Subclasses:

FederatedRunningQuery

public class ChunkedRunningQuery
extends AbstractRunningQuery

IRunningQuery implementation based on the assignment of IChunkMessage(s) to an operator task. Operators (other than those with "operator-at-once" evaluation semantics) will typically executed multiple times, consuming at least one IChunkMessage each time they are evaluated. IChunkMessages target a specific operator (bopId) and shard (shardId). In scale-out, binding sets will be mapped across the target access path and may be replicated to one or more nodes depending on the distribution of the shards. This evaluation strategy is compatible with both the Journal (aka standalone) and the IBigdataFederation (aka clustered or scale-out).

Note: The challenge with this implementation is managing the amount of data buffered on the JVM heap without introducing control structures which can result in deadlock or starvation. This has been addressed to a large extent by sharing a lock between this class and the per-operator input work queues using modified version of the JSR 166 classes. For high volume operator at once evaluation, we need to buffer the data on the native process heap using the IMemoryManager.

Field Summary

Fields inherited from class com.bigdata.bop.engine.AbstractRunningQuery

ERR_DUPLICATE_IDENTIFIER, ERR_NO_SUCH_BOP, ERR_NOT_CONTROLLER, ERR_NOT_PIPELINE_START, ERR_QUERY_DONE, lock

Constructor Summary

Constructors

Constructor and Description
ChunkedRunningQuery(QueryEngine queryEngine, UUID queryId, boolean controller, IQueryClient clientProxy, PipelineOp query, IChunkMessage<IBindingSet> realSource)

Method Summary

Methods

Modifier and Type	Method and Description
protected boolean	acceptChunk(IChunkMessage<IBindingSet> msg) Make a chunk of binding sets available for consumption by the query.
protected boolean	cancelRunningOperators(boolean mayInterruptIfRunning) Cancel any running operators for this query on this node (internal API).
protected void	consumeChunk() Instruct the IRunningQuery to consume an IChunkMessage already on its input queue.
protected IChunkHandler	getChunkHandler() Return the effective IChunkHandler for this query.
protected IChunkHandler	getChunkHandler(QueryEngine queryEngine, PipelineOp query) Factory returns the effective IChunkHandler for this query.
protected Map<Integer,QueueStats>	getQueueStats() Return a summary of the work queue for the operators in this query (non-blocking).
protected void	haltOp(IHaltOpMessage msg) Overridden to attempt to consume another chunk each time an operator reports that it has halted evaluation.
protected void	releaseAcceptedMessages() Close the IAsynchronousIterator for any IChunkMessages which have been accepted for this queue on this node (internal API).

Methods inherited from class com.bigdata.bop.engine.AbstractRunningQuery

addChild, cancel, cancelQueryOnPeers, checkDeadline, doLastPass, get, get, getAsThrownCause, getAttributes, getBOp, getBOpIndex, getCause, getChildren, getDeadline, getDoneTime, getElapsed, getFederation, getFuture, getLocalIndexManager, getMemoryManager, getQuery, getQueryBuffer, getQueryController, getQueryEngine, getQueryId, getRunningCount, getRunState, getStartedOnCount, getStartTime, getStaticAnalysisStats, getStats, getStats, halt, halt, isAtOnceReady, isCancelled, isController, isDone, isRootCauseInterrupt, iterator, newQueryBuffer, releaseNativeMemoryForOperator, releaseNativeMemoryForQuery, runStateString, setDeadline, setStaticAnalysisStats, startOp, startQuery, toString, tryGetRunState

Methods inherited from class java.lang.Object

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

Constructor Detail

ChunkedRunningQuery

public ChunkedRunningQuery(QueryEngine queryEngine,
                   UUID queryId,
                   boolean controller,
                   IQueryClient clientProxy,
                   PipelineOp query,
                   IChunkMessage<IBindingSet> realSource)

Parameters:

queryEngine - The QueryEngine on which the query is running. In scale-out, a query is typically instantiated on many QueryEngines.

queryId - The identifier for that query.

controller - true iff the QueryEngine is the query controller for this query (the QueryEngine which will coordinate the query evaluation).

clientProxy - The query controller. In standalone, this is the same as the queryEngine. In scale-out, this is a proxy for the query controller whenever the query is instantiated on a node other than the query controller itself.

query - The query.

saStats - Statistics object containing static analysis statistics

Throws:

IllegalArgumentException - if any argument is null.

IllegalArgumentException - if the readTimestamp is ITx.UNISOLATED (queries may not read on the unisolated indices).

IllegalArgumentException - if the writeTimestamp is neither ITx.UNISOLATED nor a read-write transaction identifier.

Method Detail

acceptChunk

protected boolean acceptChunk(IChunkMessage<IBindingSet> msg)

Make a chunk of binding sets available for consumption by the query.

Note: this is invoked by QueryEngine.acceptChunk(IChunkMessage)

Specified by:

acceptChunk in class AbstractRunningQuery

Parameters:

msg - The chunk.

Returns:

true if the message was accepted.

consumeChunk

protected void consumeChunk()

Instruct the IRunningQuery to consume an IChunkMessage already on its input queue..

Examines the input queue for each (bopId,partitionId). If there is work available and no task is currently running, then drain the work queue and submit a task to consume that work.

Specified by:

consumeChunk in class AbstractRunningQuery

haltOp

protected void haltOp(IHaltOpMessage msg)

Overridden to attempt to consume another chunk each time an operator reports that it has halted evaluation. This is necessary because the haltOp() message can arrive asynchronously, so we need to test the work queues in case there are "at-once" operators awaiting the termination of their predecessor(s) in the pipeline.

Overrides:

haltOp in class AbstractRunningQuery

Parameters:

msg - The IHaltOpMessage

cancelRunningOperators

protected boolean cancelRunningOperators(boolean mayInterruptIfRunning)

Description copied from class: AbstractRunningQuery

Cancel any running operators for this query on this node (internal API).

Specified by:

cancelRunningOperators in class AbstractRunningQuery

Returns:

true if any operators were cancelled.

releaseAcceptedMessages

protected void releaseAcceptedMessages()

Description copied from class: AbstractRunningQuery

Close the IAsynchronousIterator for any IChunkMessages which have been accepted for this queue on this node (internal API).

Note: This must be invoked while holding a lock which is exclusive with the lock used to hand off IChunkMessages to operator tasks otherwise we could wind up invoking IAsynchronousIterator.close() from on an IAsynchronousIterator running in a different thread. That would cause visibility problems in the close() semantics unless the IAsynchronousIterator is thread-safe for close (e.g., volatile write, synchronized, etc.). The appropriate lock for this is AbstractRunningQuery.lock. This method is only invoked out of AbstractRunningQuery.cancel(boolean) which owns that lock.

Specified by:

releaseAcceptedMessages in class AbstractRunningQuery

getQueueStats

protected Map<Integer,QueueStats> getQueueStats()

Return a summary of the work queue for the operators in this query (non-blocking).

Returns:

A map whose keys are the operator identifiers and whose values provide summary statistics for the work queue(s) for those operators.

Note: For a cluster, there is one work queue per (operator,shard) pair.

getChunkHandler

protected IChunkHandler getChunkHandler(QueryEngine queryEngine,
                            PipelineOp query)

Factory returns the effective IChunkHandler for this query.

Parameters:

queryEngine -

query -

Returns:

The effective IChunkHandler policy.

See Also:

QueryHints.QUERY_ENGINE_CHUNK_HANDLER, Vector query engine on native heap.

getChunkHandler

protected final IChunkHandler getChunkHandler()

Return the effective IChunkHandler for this query.

See Also:

Vector query engine on native heap.