NestedLoopJoinExecStream Class Reference

NestedLoopJoinExecStream performs a nested loop join between two inputs by iterating over the first input once and opening and re-iterating over the second input for each tuple from the first. More...

#include <NestedLoopJoinExecStream.h>

Inheritance diagram for NestedLoopJoinExecStream:

List of all members.


Public Member Functions
virtual void	prepare (NestedLoopJoinExecStreamParams const &params)
virtual void	open (bool restart)
	Opens this stream, acquiring any resources needed in order to be able to fetch data.
virtual void	prepare (CartesianJoinExecStreamParams const &params)
virtual void	prepare (ConfluenceExecStreamParams const &params)
virtual void	prepare (SingleOutputExecStreamParams const &params)
virtual void	prepare (ExecStreamParams const &params)
	Prepares this stream for execution.
virtual ExecStreamResult	execute (ExecStreamQuantum const &quantum)
	Executes this stream.
virtual void	setInputBufAccessors (std::vector< SharedExecStreamBufAccessor > const &inAccessors)
	Initializes the buffer accessors for inputs to this stream.
virtual ExecStreamBufProvision	getInputBufProvision () const
	Queries the BufferProvision which this stream requires of its inputs when consuming their tuples.
virtual void	setOutputBufAccessors (std::vector< SharedExecStreamBufAccessor > const &outAccessors)
	Initializes the buffer accessors for outputs from this stream.
virtual ExecStreamBufProvision	getOutputBufProvision () const
	Queries the BufferProvision which this stream is capable of when producing tuples.
virtual bool	canEarlyClose ()
	Returns: true if the stream can be closed early
ExecStreamGraph &	getGraph () const
	Returns: reference to containing graph
ExecStreamId	getStreamId () const
	Returns: the identifier for this stream within containing graph
virtual void	getResourceRequirements (ExecStreamResourceQuantity &minQuantity, ExecStreamResourceQuantity &optQuantity, ExecStreamResourceSettingType &optType)
	Determines resource requirements for this stream.
virtual void	getResourceRequirements (ExecStreamResourceQuantity &minQuantity, ExecStreamResourceQuantity &optQuantity)
virtual void	setResourceAllocation (ExecStreamResourceQuantity &quantity)
	Sets current resource allocation for this stream.
virtual void	setName (std::string const &)
	Sets unique name of this stream.
virtual std::string const &	getName () const
	Returns: the name of this stream, as known by the optimizer
virtual bool	mayBlock () const
	Queries whether this stream's implementation may block when execute() is called.
virtual void	checkAbort () const
	Checks whether there is an abort request for this stream's scheduler.
virtual ExecStreamBufProvision	getOutputBufConversion () const
	Queries the BufferProvision to which this stream needs its output to be converted, if any.
bool	isClosed () const
	Returns: whether the object has been closed
void	close ()
	Closes this object, releasing any unallocated resources.
virtual void	initTraceSource (SharedTraceTarget pTraceTarget, std::string name)
	For use when initialization has to be deferred until after construction.
void	trace (TraceLevel level, std::string message) const
	Records a trace message.
bool	isTracing () const
	Returns: true iff tracing is enabled for this source
bool	isTracingLevel (TraceLevel level) const
	Determines whether a particular level is being traced.
TraceTarget &	getTraceTarget () const
	Returns: the TraceTarget for this source
SharedTraceTarget	getSharedTraceTarget () const
	Returns: the SharedTraceTarget for this source
std::string	getTraceSourceName () const
	Gets the name of this source.
void	setTraceSourceName (std::string const &n)
	Sets the name of this source.
TraceLevel	getMinimumTraceLevel () const
void	disableTracing ()
virtual void	initErrorSource (SharedErrorTarget pErrorTarget, const std::string &name)
	For use when initialization has to be deferred until after construction.
void	postError (ErrorLevel level, const std::string &message, void *address, long capacity, int index)
	Posts an exception, such as a row exception.
void	postError (ErrorLevel level, const std::string &message, const TupleDescriptor &errorDesc, const TupleData &errorTuple, int index)
	Posts an exception, such as a row exception.
bool	hasTarget () const
	Returns: true iff an error target has been set
ErrorTarget &	getErrorTarget () const
	Returns: the ErrorTarget for this source
SharedErrorTarget	getSharedErrorTarget () const
	Returns: the SharedErrorTarget for this source
std::string	getErrorSourceName () const
	Gets the name of this source.
void	setErrorSourceName (std::string const &n)
	Sets the name of this source.
void	disableTarget ()
Protected Member Functions
virtual void	closeImpl ()
	Implements ClosableObject.
Protected Attributes
bool	leftOuter
bool	rightInputEmpty
TupleData	outputData
SharedExecStreamBufAccessor	pLeftBufAccessor
SharedExecStreamBufAccessor	pRightBufAccessor
SharedExecStream	pRightInput
uint	nLeftAttributes
std::vector< SharedExecStreamBufAccessor >	inAccessors
SharedExecStreamBufAccessor	pOutAccessor
bool	isOpen
	Whether this stream is currently open.
ExecStreamGraph *	pGraph
	Dataflow graph containing this stream.
ExecStreamId	id
	Identifier for this stream; local to its containing graph.
std::string	name
	Name of stream, as known by optimizer.
SharedDynamicParamManager	pDynamicParamManager
	The dynamic parameter manager available to this stream.
SharedLogicalTxn	pTxn
	The transaction embracing the stream.
ExecStreamResourceQuantity	resourceAllocation
	Resource quantities currently allocated to this stream.
SharedCacheAccessor	pQuotaAccessor
	CacheAccessor used for quota tracking.
SharedCacheAccessor	pScratchQuotaAccessor
	CacheAccessor used for scratch page quota tracking.
bool	needsClose
Private Member Functions
virtual bool	checkNumInputs ()
	Returns: true if the number of inputs to the stream is correct
virtual ExecStreamResult	preProcessRightInput ()
	Creates temporary index used in nested loop join.
virtual void	processLeftInput ()
	Passes join keys from the left input to the right input using dynamic parameters.
Private Attributes
bool	preProcessingDone
	True if pre-processing on third input completed.
std::vector< NestedLoopJoinKey >	leftJoinKeys
	Dynamic parameters corresponding to the left join keys.

Detailed Description

NestedLoopJoinExecStream performs a nested loop join between two inputs by iterating over the first input once and opening and re-iterating over the second input for each tuple from the first.

Join keys from the first (left) input are passed to the second (right) input through dynamic parameters. An optional third input will do any pre-processing required prior to executing the actual nested loop join.

NOTE: The input that does pre-processing needs to be the third input because it may need to be opened before other streams in the overall stream graph. Due to the reverse topological sort order in which a stream graph is opened, the last input into a stream is opened before any of the other inputs.

Author:: Zelaine Fong

Version:

Id: //open/dev/fennel/exec/NestedLoopJoinExecStream.h#6

Definition at line 75 of file NestedLoopJoinExecStream.h.

Member Function Documentation

bool NestedLoopJoinExecStream::checkNumInputs ( ) [private, virtual]

Returns:: true if the number of inputs to the stream is correct

Reimplemented from CartesianJoinExecStream.

Definition at line 39 of file NestedLoopJoinExecStream.cpp.

References ConfluenceExecStream::inAccessors.

00040 {
00041     return (inAccessors.size() >= 2 && inAccessors.size() <= 3);
00042 }

ExecStreamResult NestedLoopJoinExecStream::preProcessRightInput ( ) [private, virtual]

Creates temporary index used in nested loop join.

Returns:: EXECRC_BUF_UNDERFLOW if request to create temporary index hasn't been initiated yet

Reimplemented from CartesianJoinExecStream.

Definition at line 64 of file NestedLoopJoinExecStream.cpp.

References EXECBUF_EOS, EXECRC_BUF_UNDERFLOW, EXECRC_YIELD, ConfluenceExecStream::inAccessors, and preProcessingDone.

00065 {
00066     // Create the temporary index by requesting production on the 3rd input
00067     if (!preProcessingDone && inAccessors.size() == 3) {
00068         if (inAccessors[2]->getState() != EXECBUF_EOS) {
00069             inAccessors[2]->requestProduction();
00070             return EXECRC_BUF_UNDERFLOW;
00071         }
00072     }
00073     preProcessingDone = true;
00074     return EXECRC_YIELD;
00075 }

void NestedLoopJoinExecStream::processLeftInput ( ) [private, virtual]

Passes join keys from the left input to the right input using dynamic parameters.

Reimplemented from CartesianJoinExecStream.

Definition at line 77 of file NestedLoopJoinExecStream.cpp.

References leftJoinKeys, CartesianJoinExecStream::outputData, and ExecStream::pDynamicParamManager.

00078 {
00079     std::vector<NestedLoopJoinKey>::iterator it;
00080     for (it = leftJoinKeys.begin(); it != leftJoinKeys.end(); it++) {
00081         pDynamicParamManager->writeParam(
00082             it->dynamicParamId,
00083             outputData[it->leftAttributeOrdinal]);
00084     }
00085 }

void NestedLoopJoinExecStream::prepare ( NestedLoopJoinExecStreamParams const & params ) [virtual]

Definition at line 30 of file NestedLoopJoinExecStream.cpp.

References NestedLoopJoinExecStreamParams::leftJoinKeys, leftJoinKeys, CartesianJoinExecStream::nLeftAttributes, and CartesianJoinExecStream::prepare().

00032 {
00033     CartesianJoinExecStream::prepare(params);
00034 
00035     leftJoinKeys.assign(params.leftJoinKeys.begin(), params.leftJoinKeys.end());
00036     assert(leftJoinKeys.size() <= nLeftAttributes);
00037 }

void NestedLoopJoinExecStream::open ( bool restart ) [virtual]

Opens this stream, acquiring any resources needed in order to be able to fetch data.

A precondition is that input streams must already be opened. A stream can be closed and reopened.

Parameters:

restart

if true, the stream must be already open, and should reset itself to start from the beginning of its result set

Reimplemented from CartesianJoinExecStream.

Definition at line 44 of file NestedLoopJoinExecStream.cpp.

References leftJoinKeys, CartesianJoinExecStream::open(), ExecStream::pDynamicParamManager, CartesianJoinExecStream::pLeftBufAccessor, and preProcessingDone.

00045 {
00046     CartesianJoinExecStream::open(restart);
00047 
00048     if (!restart) {
00049         std::vector<NestedLoopJoinKey>::iterator it;
00050         for (it = leftJoinKeys.begin(); it != leftJoinKeys.end(); it++) {
00051             pDynamicParamManager->createParam(
00052                 it->dynamicParamId,
00053                 pLeftBufAccessor->getTupleDesc()[it->leftAttributeOrdinal]);
00054         }
00055 
00056         // Initialize this here and don't reset on restarts, since the
00057         // defined behavior is that the pre-processing is only done once
00058         // per stream graph execution, even if the stream is re-opened in
00059         // restart mode
00060         preProcessingDone = false;
00061     }
00062 }

void CartesianJoinExecStream::prepare ( CartesianJoinExecStreamParams const & params ) [virtual, inherited]

Definition at line 33 of file CartesianJoinExecStream.cpp.

Referenced by prepare().

00035 {
00036     assert(checkNumInputs());
00037     pLeftBufAccessor = inAccessors[0];
00038     assert(pLeftBufAccessor);
00039 
00040     pRightBufAccessor = inAccessors[1];
00041     assert(pRightBufAccessor);
00042 
00043     leftOuter = params.leftOuter;
00044 
00045     SharedExecStream pLeftInput = pGraph->getStreamInput(getStreamId(), 0);
00046     assert(pLeftInput);
00047     pRightInput = pGraph->getStreamInput(getStreamId(), 1);
00048     assert(pRightInput);
00049     FENNEL_TRACE(
00050         TRACE_FINE,
00051         "left input " << pLeftInput->getStreamId() <<
00052         ' ' << pLeftInput->getName() <<
00053         ", right input " << pRightInput->getStreamId() <<
00054         ' ' << pRightInput->getName());
00055 
00056 
00057     TupleDescriptor const &leftDesc = pLeftBufAccessor->getTupleDesc();
00058     TupleDescriptor const &rightDesc = pRightBufAccessor->getTupleDesc();
00059 
00060     TupleDescriptor outputDesc;
00061     outputDesc.insert(outputDesc.end(),leftDesc.begin(),leftDesc.end());
00062     uint iFirstRight = outputDesc.size();
00063     outputDesc.insert(outputDesc.end(),rightDesc.begin(),rightDesc.end());
00064     if (leftOuter) {
00065         // Right side is null-generating; have to adjust tuple descriptor
00066         // accordingly.
00067         for (uint i = iFirstRight; i < outputDesc.size(); ++i) {
00068             outputDesc[i].isNullable = true;
00069         }
00070     }
00071     if (params.outputTupleDesc.size()) {
00072         assert(params.outputTupleDesc == outputDesc);
00073     }
00074     outputData.compute(outputDesc);
00075     pOutAccessor->setTupleShape(outputDesc);
00076 
00077     nLeftAttributes = leftDesc.size();
00078 
00079     ConfluenceExecStream::prepare(params);
00080 }

void ConfluenceExecStream::prepare ( ConfluenceExecStreamParams const & params ) [virtual, inherited]

Definition at line 37 of file ConfluenceExecStream.cpp.

References ConfluenceExecStream::getInputBufProvision(), ConfluenceExecStream::inAccessors, and SingleOutputExecStream::prepare().

Referenced by LcsRowScanBaseExecStream::prepare(), LbmUnionExecStream::prepare(), LbmChopperExecStream::prepare(), LbmBitOpExecStream::prepare(), LhxJoinExecStream::prepare(), MergeExecStream::prepare(), CorrelationJoinExecStream::prepare(), CartesianJoinExecStream::prepare(), and BarrierExecStream::prepare().

00038 {
00039     SingleOutputExecStream::prepare(params);
00040 
00041     for (uint i = 0; i < inAccessors.size(); ++i) {
00042         assert(inAccessors[i]->getProvision() == getInputBufProvision());
00043     }
00044 }

void SingleOutputExecStream::prepare ( SingleOutputExecStreamParams const & params ) [virtual, inherited]

Definition at line 48 of file SingleOutputExecStream.cpp.

References SingleOutputExecStream::getOutputBufProvision(), SingleOutputExecStreamParams::outputTupleDesc, SingleOutputExecStreamParams::outputTupleFormat, SingleOutputExecStream::pOutAccessor, and ExecStream::prepare().

Referenced by BTreeExecStream::prepare(), FlatFileExecStreamImpl::prepare(), ValuesExecStream::prepare(), MockResourceExecStream::prepare(), MockProducerExecStream::prepare(), ConfluenceExecStream::prepare(), and ConduitExecStream::prepare().

00049 {
00050     ExecStream::prepare(params);
00051     assert(pOutAccessor);
00052     assert(pOutAccessor->getProvision() == getOutputBufProvision());
00053     if (pOutAccessor->getTupleDesc().empty()) {
00054         assert(!params.outputTupleDesc.empty());
00055         pOutAccessor->setTupleShape(
00056             params.outputTupleDesc,
00057             params.outputTupleFormat);
00058     }
00059 }

void ExecStream::prepare ( ExecStreamParams const & params ) [virtual, inherited]

Prepares this stream for execution.

A precondition is that input streams must already be defined and prepared. As an effect of this call, the tuple shape should be defined for all output buffers and remain unchanged for the lifetime of the stream. This method is only ever called once, before the first open. Although this method is virtual, derived classes may choose to define an overloaded version instead with a specialized covariant parameter class.

Parameters:

params

instance of stream parameterization class which should be used to prepare this stream

Definition at line 84 of file ExecStream.cpp.

References ExecStreamGraph::getDynamicParamManager(), SegmentAccessor::pCacheAccessor, ExecStreamParams::pCacheAccessor, ExecStream::pDynamicParamManager, ExecStream::pGraph, ExecStream::pQuotaAccessor, ExecStream::pScratchQuotaAccessor, and ExecStreamParams::scratchAccessor.

Referenced by JavaTransformExecStream::prepare(), SingleOutputExecStream::prepare(), and SingleInputExecStream::prepare().

00085 {
00086     if (pGraph) {
00087         pDynamicParamManager = pGraph->getDynamicParamManager();
00088     }
00089     pQuotaAccessor = params.pCacheAccessor;
00090     pScratchQuotaAccessor = params.scratchAccessor.pCacheAccessor;
00091 }

ExecStreamResult CartesianJoinExecStream::execute ( ExecStreamQuantum const & quantum ) [virtual, inherited]

Executes this stream.

Parameters:

quantum

governs the maximum amount of execution to perform

Returns:: code indicating reason execution ceased

Implements ExecStream.

Definition at line 103 of file CartesianJoinExecStream.cpp.

00105 {
00106     // TODO:  lots of small optimizations possible here
00107 
00108     // TODO jvs 6-Nov-2004: one big optimization would be to perform
00109     // buffer-to-buffer joins instead of row-to-buffer joins.  This would
00110     // reduce the number of times the right input needs to be iterated by the
00111     // average number of rows in a buffer from the left input.  However,  the
00112     // output ordering would also be affected, so we might want to provide a
00113     // parameter to control this behavior.
00114 
00115     // Also, for outer join, once we know the right input is empty, it's always
00116     // going to be empty for every left row, so we could just stop trying to
00117     // re-execute the right hand side.
00118 
00119     uint nTuplesProduced = 0;
00120 
00121     for (;;) {
00122         if (!pLeftBufAccessor->isTupleConsumptionPending()) {
00123             if (pLeftBufAccessor->getState() == EXECBUF_EOS) {
00124                 pOutAccessor->markEOS();
00125                 return EXECRC_EOS;
00126             }
00127             if (!pLeftBufAccessor->demandData()) {
00128                 FENNEL_TRACE_THREAD(
00129                     TRACE_FINE,
00130                     "left underflow; left input " << pLeftBufAccessor <<
00131                     " right input " << pRightBufAccessor);
00132                 return EXECRC_BUF_UNDERFLOW;
00133             }
00134             pLeftBufAccessor->unmarshalTuple(outputData);
00135             processLeftInput();
00136             rightInputEmpty = true;
00137         }
00138         ExecStreamResult rc = preProcessRightInput();
00139         if (rc != EXECRC_YIELD) {
00140             return rc;
00141         }
00142         for (;;) {
00143             if (!pRightBufAccessor->isTupleConsumptionPending()) {
00144                 if (pRightBufAccessor->getState() == EXECBUF_EOS) {
00145                     if (leftOuter && rightInputEmpty) {
00146                         // put null in outputdata to the right of
00147                         // nLeftAttributes
00148                         for (int i = nLeftAttributes;
00149                              i < outputData.size(); ++i)
00150                         {
00151                             outputData[i].pData = NULL;
00152                         }
00153 
00154                         if (pOutAccessor->produceTuple(outputData)) {
00155                             ++nTuplesProduced;
00156                         } else {
00157                             return EXECRC_BUF_OVERFLOW;
00158                         }
00159 
00160                         if (nTuplesProduced >= quantum.nTuplesMax) {
00161                             return EXECRC_QUANTUM_EXPIRED;
00162                         }
00163                     }
00164 
00165                     pLeftBufAccessor->consumeTuple();
00166                     // restart right input stream
00167                     pRightInput->open(true);
00168                     FENNEL_TRACE_THREAD(
00169                         TRACE_FINE,
00170                         "re-opened right input " << pRightBufAccessor);
00171                     // NOTE: break out of the inner for loop, which will take
00172                     // us back to the top of the outer for loop
00173                     break;
00174                 }
00175                 if (!pRightBufAccessor->demandData()) {
00176                     FENNEL_TRACE_THREAD(
00177                         TRACE_FINE,
00178                         "right underflow; left input " << pLeftBufAccessor <<
00179                         " right input " << pRightBufAccessor);
00180                     return EXECRC_BUF_UNDERFLOW;
00181                 }
00182                 rightInputEmpty = false;
00183                 pRightBufAccessor->unmarshalTuple(
00184                     outputData, nLeftAttributes);
00185                 break;
00186             }
00187 
00188             if (pOutAccessor->produceTuple(outputData)) {
00189                 ++nTuplesProduced;
00190             } else {
00191                 return EXECRC_BUF_OVERFLOW;
00192             }
00193 
00194             pRightBufAccessor->consumeTuple();
00195 
00196             if (nTuplesProduced >= quantum.nTuplesMax) {
00197                 return EXECRC_QUANTUM_EXPIRED;
00198             }
00199         }
00200     }
00201 }

void ConfluenceExecStream::setInputBufAccessors ( std::vector< SharedExecStreamBufAccessor > const & inAccessors ) [virtual, inherited]

Initializes the buffer accessors for inputs to this stream.

This method is only ever called once, before prepare.

Parameters:

inAccessors

buffer accessors ordered by input stream

Reimplemented from SingleOutputExecStream.

Definition at line 31 of file ConfluenceExecStream.cpp.

References ConfluenceExecStream::inAccessors.

00033 {
00034     inAccessors = inAccessorsInit;
00035 }

ExecStreamBufProvision ConfluenceExecStream::getInputBufProvision ( ) const [virtual, inherited]

Queries the BufferProvision which this stream requires of its inputs when consuming their tuples.

Returns:: required model; default is BUFPROV_NONE

Reimplemented from ExecStream.

Definition at line 58 of file ConfluenceExecStream.cpp.

References BUFPROV_PRODUCER.

Referenced by ConfluenceExecStream::prepare().

00059 {
00060     return BUFPROV_PRODUCER;
00061 }

void SingleOutputExecStream::setOutputBufAccessors ( std::vector< SharedExecStreamBufAccessor > const & outAccessors ) [virtual, inherited]

Initializes the buffer accessors for outputs from this stream.

This method is only ever called once, before prepare.

Parameters:

outAccessors

buffer accessors ordered by output stream

Implements ExecStream.

Reimplemented in ConduitExecStream.

Definition at line 41 of file SingleOutputExecStream.cpp.

References SingleOutputExecStream::pOutAccessor.

Referenced by ConduitExecStream::setOutputBufAccessors().

00043 {
00044     assert(outAccessors.size() == 1);
00045     pOutAccessor = outAccessors[0];
00046 }

ExecStreamBufProvision SingleOutputExecStream::getOutputBufProvision ( ) const [virtual, inherited]

Queries the BufferProvision which this stream is capable of when producing tuples.

Returns:: supported model; default is BUFPROV_NONE

Reimplemented from ExecStream.

Reimplemented in BarrierExecStream, DoubleBufferExecStream, MergeExecStream, MockResourceExecStream, ScratchBufferExecStream, SegBufferExecStream, SegBufferReaderExecStream, ValuesExecStream, FtrsTableWriterExecStream, and LcsClusterAppendExecStream.

Definition at line 69 of file SingleOutputExecStream.cpp.

References BUFPROV_CONSUMER.

Referenced by SingleOutputExecStream::prepare().

00070 {
00071     return BUFPROV_CONSUMER;
00072 }

void ExecStream::closeImpl ( ) [protected, virtual, inherited]

Implements ClosableObject.

ExecStream implementations may override this to release any resources acquired while open.

Implements ClosableObject.

Reimplemented in BarrierExecStream, DoubleBufferExecStream, MockResourceExecStream, ReshapeExecStream, ScratchBufferExecStream, SegBufferExecStream, SegBufferReaderExecStream, SegBufferWriterExecStream, JavaSinkExecStream, JavaTransformExecStream, FlatFileExecStreamImpl, BTreeExecStream, BTreeInsertExecStream, BTreePrefetchSearchExecStream, BTreeReadExecStream, BTreeSearchExecStream, FtrsTableWriterExecStream, LhxAggExecStream, LhxJoinExecStream, LbmBitOpExecStream, LbmChopperExecStream, LbmGeneratorExecStream, LbmIntersectExecStream, LbmMinusExecStream, LbmSplicerExecStream, LbmUnionExecStream, LcsClusterAppendExecStream, LcsRowScanBaseExecStream, LcsRowScanExecStream, and ExternalSortExecStreamImpl.

Definition at line 54 of file ExecStream.cpp.

References ExecStream::isOpen.

00055 {
00056     isOpen = false;
00057 
00058     // REVIEW jvs 19-July-2004:  It would be nice to be able to do this, making
00059     // sure no cache access is attempted while stream is closed.  However,
00060     // it currently causes trouble with TableWriters, which need
00061     // cache access for txn replay.
00062     /*
00063     if (pQuotaAccessor) {
00064         pQuotaAccessor->setMaxLockedPages(0);
00065     }
00066     if (pScratchQuotaAccessor) {
00067         pScratchQuotaAccessor->setMaxLockedPages(0);
00068     }
00069     */
00070 }

bool ExecStream::canEarlyClose ( ) [virtual, inherited]

Returns:: true if the stream can be closed early

Reimplemented in SegBufferWriterExecStream.

Definition at line 49 of file ExecStream.cpp.

00050 {
00051     return true;
00052 }

ExecStreamGraph & ExecStream::getGraph ( ) const [inline, inherited]

Returns:: reference to containing graph

Definition at line 293 of file ExecStream.h.

References ExecStream::pGraph.

Referenced by ExternalSortExecStreamImpl::execute(), JavaSinkExecStream::execute(), SegBufferWriterExecStream::execute(), SegBufferExecStream::execute(), BarrierExecStream::execute(), Java_net_sf_farrago_fennel_FennelStorage_tupleStreamFetch(), Java_net_sf_farrago_fennel_FennelStorage_tupleStreamTransformFetch(), JavaSinkExecStream::stuffByteBuffer(), and ExecStreamScheduler::traceStreamBuffers().

00294 {
00295     assert(pGraph);
00296     return *pGraph;
00297 }

ExecStreamId ExecStream::getStreamId ( ) const [inline, inherited]

Returns:: the identifier for this stream within containing graph

Definition at line 288 of file ExecStream.h.

References ExecStream::id.

Referenced by ExternalSortExecStreamImpl::execute(), SegBufferWriterExecStream::execute(), SegBufferExecStream::execute(), CorrelationJoinExecStream::execute(), BarrierExecStream::execute(), Java_net_sf_farrago_fennel_FennelStorage_tupleStreamTransformFetch(), JavaTransformExecStream::open(), SingleInputExecStream::open(), ConfluenceExecStream::open(), CartesianJoinExecStream::prepare(), ParallelExecStreamScheduler::readStream(), DfsTreeExecStreamScheduler::readStream(), LbmMinusExecStream::restartSubtrahends(), ExecStreamScheduler::tracePostExecution(), ExecStreamScheduler::tracePreExecution(), ExecStreamScheduler::traceStreamBuffers(), and ParallelExecStreamScheduler::tryExecuteTask().

00289 {
00290     return id;
00291 }

void ExecStream::getResourceRequirements	(	ExecStreamResourceQuantity &	minQuantity,
		ExecStreamResourceQuantity &	optQuantity,
		ExecStreamResourceSettingType &	optType
	)			`[virtual, inherited]`

Determines resource requirements for this stream.

Default implementation declares zero resource requirements.

Parameters:

	minQuantity	receives the minimum resource quantity needed by this stream in order to execute
	optQuantity	receives the resource quantity needed by this stream in order to execute optimally
	optType	Receives the value indicating the accuracy of the optQuantity parameter. This parameter is optional and defaults to EXEC_RESOURCE_ACCURATE if omitted. If the optimum setting is an estimate or no value can be specified (e.g., due to lack of statistics), then this parameter needs to be used to indicate a non-accurate optimum resource setting.

Reimplemented in MockResourceExecStream, BTreePrefetchSearchExecStream, LhxAggExecStream, LhxJoinExecStream, LbmGeneratorExecStream, LbmUnionExecStream, and ExternalSortExecStreamImpl.

Definition at line 93 of file ExecStream.cpp.

References EXEC_RESOURCE_ACCURATE.

00097 {
00098     getResourceRequirements(minQuantity, optQuantity);
00099     optType = EXEC_RESOURCE_ACCURATE;
00100 }

void ExecStream::getResourceRequirements	(	ExecStreamResourceQuantity &	minQuantity,
		ExecStreamResourceQuantity &	optQuantity
	)			`[virtual, inherited]`

Reimplemented in DoubleBufferExecStream, ScratchBufferExecStream, SegBufferExecStream, SegBufferReaderExecStream, SegBufferWriterExecStream, FlatFileExecStreamImpl, BTreeInsertExecStream, BTreeReadExecStream, FtrsTableWriterExecStream, LbmChopperExecStream, LbmSplicerExecStream, LcsClusterAppendExecStream, LcsClusterReplaceExecStream, LcsRowScanBaseExecStream, and LcsRowScanExecStream.

Definition at line 102 of file ExecStream.cpp.

References ExecStreamResourceQuantity::nCachePages, and ExecStreamResourceQuantity::nThreads.

00105 {
00106     minQuantity.nThreads = 0;
00107     minQuantity.nCachePages = 0;
00108     optQuantity = minQuantity;
00109 }

void ExecStream::setResourceAllocation ( ExecStreamResourceQuantity & quantity ) [virtual, inherited]

Sets current resource allocation for this stream.

If called while the stream is open, this indicates a request for the stream to dynamically adjust its memory usage. If the stream is incapable of honoring the request, it should update quantity with the actual amounts still in use.

Parameters:

quantity

allocated resource quantity

Reimplemented in MockResourceExecStream, BTreePrefetchSearchExecStream, LhxAggExecStream, LhxJoinExecStream, LbmGeneratorExecStream, LbmUnionExecStream, and ExternalSortExecStreamImpl.

Definition at line 111 of file ExecStream.cpp.

References ExecStreamResourceQuantity::nCachePages, ExecStream::pQuotaAccessor, ExecStream::pScratchQuotaAccessor, and ExecStream::resourceAllocation.

Referenced by ExternalSortExecStreamImpl::setResourceAllocation(), LbmUnionExecStream::setResourceAllocation(), LbmGeneratorExecStream::setResourceAllocation(), LhxJoinExecStream::setResourceAllocation(), LhxAggExecStream::setResourceAllocation(), and BTreePrefetchSearchExecStream::setResourceAllocation().

00113 {
00114     resourceAllocation = quantity;
00115     if (pQuotaAccessor) {
00116         pQuotaAccessor->setMaxLockedPages(quantity.nCachePages);
00117     }
00118     if (pScratchQuotaAccessor) {
00119         pScratchQuotaAccessor->setMaxLockedPages(quantity.nCachePages);
00120     }
00121 }

void ExecStream::setName ( std::string const & ) [virtual, inherited]

Sets unique name of this stream.

Definition at line 157 of file ExecStream.cpp.

References ExecStream::name.

00158 {
00159     name = nameInit;
00160 }

std::string const & ExecStream::getName ( ) const [virtual, inherited]

Returns:: the name of this stream, as known by the optimizer

Definition at line 162 of file ExecStream.cpp.

References ExecStream::name.

Referenced by DfsTreeExecStreamScheduler::findNextConsumer(), ParallelExecStreamScheduler::readStream(), DfsTreeExecStreamScheduler::readStream(), ExecStreamScheduler::tracePostExecution(), and ExecStreamScheduler::tracePreExecution().

00163 {
00164     return name;
00165 }

bool ExecStream::mayBlock ( ) const [virtual, inherited]

Queries whether this stream's implementation may block when execute() is called.

For accurate scheduling, non-blocking implementations are preferred; the scheduler must be aware of the potential for blocking so that it can allocate extra threads accordingly.

Returns:: whether stream may block; default is false

Definition at line 167 of file ExecStream.cpp.

00168 {
00169     return false;
00170 }

void ExecStream::checkAbort ( ) const [virtual, inherited]

Checks whether there is an abort request for this stream's scheduler.

Normally, streams don't need to check this, since the scheduler services abort requests in between quanta. However, streams which enter long-running loops need to check for themselves. If an abort is scheduled, this method will throw an AbortExcn automatically.

Definition at line 72 of file ExecStream.cpp.

References ExecStreamScheduler::checkAbort(), ExecStreamGraph::getScheduler(), and ExecStream::pGraph.

Referenced by LhxJoinExecStream::execute(), LhxAggExecStream::execute(), ExternalSortRunAccessor::fetch(), and ExternalSortMerger::fetch().

00073 {
00074     if (!pGraph) {
00075         return;
00076     }
00077     ExecStreamScheduler *pScheduler = pGraph->getScheduler();
00078     if (!pScheduler) {
00079         return;
00080     }
00081     pScheduler->checkAbort();
00082 }

ExecStreamBufProvision ExecStream::getOutputBufConversion ( ) const [virtual, inherited]

Queries the BufferProvision to which this stream needs its output to be converted, if any.

Returns:: required conversion; default is BUFPROV_NONE

Reimplemented in JavaTransformExecStream.

Definition at line 177 of file ExecStream.cpp.

References BUFPROV_NONE.

00178 {
00179     return BUFPROV_NONE;
00180 }

bool ClosableObject::isClosed ( ) const [inline, inherited]

Returns:: whether the object has been closed

Definition at line 58 of file ClosableObject.h.

00059     {
00060         return !needsClose;
00061     }

void ClosableObject::close ( ) [inherited]

Closes this object, releasing any unallocated resources.

Reimplemented in CollectExecStream, CorrelationJoinExecStream, LcsClusterAppendExecStream, and LcsClusterReplaceExecStream.

Definition at line 39 of file ClosableObject.cpp.

References ClosableObject::closeImpl(), and ClosableObject::needsClose.

Referenced by CacheImpl< PageT, VictimPolicyT >::allocatePages(), LcsRowScanBaseExecStream::closeImpl(), ExecStreamGraphImpl::closeImpl(), FlatFileBuffer::open(), ClosableObjectDestructor::operator()(), and Segment::~Segment().

00040 {
00041     if (!needsClose) {
00042         return;
00043     }
00044     needsClose = false;
00045     closeImpl();
00046 }

void TraceSource::initTraceSource	(	SharedTraceTarget	pTraceTarget,
		std::string	name
	)			`[virtual, inherited]`

For use when initialization has to be deferred until after construction.

Parameters:

	pTraceTarget	the TraceTarget to which messages will be sent
	name	the name of this source

Definition at line 46 of file TraceSource.cpp.

References TraceSource::isTracing(), TraceSource::minimumLevel, TraceSource::name, TraceSource::pTraceTarget, and TRACE_OFF.

Referenced by TestBase::beforeTestCase(), TestBase::TestBase(), and TraceSource::TraceSource().

00049 {
00050     assert(!pTraceTarget.get());
00051 
00052     pTraceTarget = pTraceTargetInit;
00053     name = nameInit;
00054     if (isTracing()) {
00055         minimumLevel = pTraceTarget->getSourceTraceLevel(name);
00056     } else {
00057         minimumLevel = TRACE_OFF;
00058     }
00059 }

void TraceSource::trace	(	TraceLevel	level,
		std::string	message
	)			const `[inherited]`

Records a trace message.

Normally only called via FENNEL_TRACE.

Parameters:

	level	severity level of event being trace
	message	the text of the message

Definition at line 61 of file TraceSource.cpp.

References TraceSource::getTraceTarget(), TraceSource::isTracing(), TraceSource::name, and TraceTarget::notifyTrace().

Referenced by Calculator::exec(), and ExecStreamScheduler::traceStreamBufferContents().

00062 {
00063     if (isTracing()) {
00064         getTraceTarget().notifyTrace(name,level,message);
00065     }
00066 }

bool TraceSource::isTracing ( ) const [inline, inherited]

Returns:: true iff tracing is enabled for this source

Definition at line 88 of file TraceSource.h.

Referenced by TraceSource::initTraceSource(), CalcExecStream::prepare(), and TraceSource::trace().

00089     {
00090         return pTraceTarget.get() ? true : false;
00091     }

bool TraceSource::isTracingLevel ( TraceLevel level ) const [inline, inherited]

Determines whether a particular level is being traced.

Parameters:

level

trace level to test

Returns:: true iff tracing is enabled for the given level

Definition at line 100 of file TraceSource.h.

Referenced by ExecStreamScheduler::addGraph(), SimpleExecStreamGovernor::assignCachePages(), SimpleExecStreamGovernor::distributeCachePages(), Calculator::exec(), ExecStreamScheduler::ExecStreamScheduler(), LcsClusterNodeWriter::getLastClusterPageForWrite(), LcsClusterNodeWriter::moveFromTempToIndex(), JavaSinkExecStream::stuffByteBuffer(), and ExecStreamScheduler::traceStreamBuffers().

00101     {
00102         return level >= minimumLevel;
00103     }

TraceTarget& TraceSource::getTraceTarget ( ) const [inline, inherited]

Returns:: the TraceTarget for this source

Definition at line 108 of file TraceSource.h.

Referenced by TraceSource::trace().

00109     {
00110         assert(isTracing());
00111         return *(pTraceTarget.get());
00112     }

SharedTraceTarget TraceSource::getSharedTraceTarget ( ) const [inline, inherited]

Returns:: the SharedTraceTarget for this source

Definition at line 117 of file TraceSource.h.

Referenced by Database::init(), LcsClusterAppendExecStream::initLoad(), and CalcExecStream::prepare().

00118     {
00119         return pTraceTarget;
00120     }

std::string TraceSource::getTraceSourceName ( ) const [inline, inherited]

Gets the name of this source.

Useful to construct nested names for subcomponents that are also TraceSources.

Returns:: the name

Definition at line 127 of file TraceSource.h.

Referenced by LcsClusterAppendExecStream::initLoad().

00128     {
00129         return name;
00130     }

void TraceSource::setTraceSourceName ( std::string const & n ) [inline, inherited]

Sets the name of this source.

Useful to construct dynamic names for fine-grained filtering.

Definition at line 136 of file TraceSource.h.

00137     {
00138         name = n;
00139     }

TraceLevel TraceSource::getMinimumTraceLevel ( ) const [inline, inherited]

Definition at line 141 of file TraceSource.h.

00142     {
00143         return minimumLevel;
00144     }

void TraceSource::disableTracing ( ) [inherited]

Definition at line 68 of file TraceSource.cpp.

References TraceSource::minimumLevel, TraceSource::pTraceTarget, and TRACE_OFF.

Referenced by TestBase::afterTestCase().

00069 {
00070     pTraceTarget.reset();
00071     minimumLevel = TRACE_OFF;
00072 }

void ErrorSource::initErrorSource	(	SharedErrorTarget	pErrorTarget,
		const std::string &	name
	)			`[virtual, inherited]`

For use when initialization has to be deferred until after construction.

Parameters:

	pErrorTarget	the ErrorTarget to which errors will be posted
	name	the name of this source

Definition at line 47 of file ErrorSource.cpp.

References ErrorSource::name, and ErrorSource::pErrorTarget.

Referenced by ErrorSource::ErrorSource().

00050 {
00051     pErrorTarget = pErrorTargetInit;
00052     name = nameInit;
00053 }

void ErrorSource::postError	(	ErrorLevel	level,
		const std::string &	message,
		void *	address,
		long	capacity,
		int	index
	)			`[inherited]`

Posts an exception, such as a row exception.

See also:: ErrorTarget for a description of the parameters

Definition at line 55 of file ErrorSource.cpp.

References ErrorSource::getErrorTarget(), ErrorSource::hasTarget(), ErrorSource::name, and ErrorTarget::notifyError().

Referenced by FlatFileExecStreamImpl::logError(), ErrorSource::postError(), and LbmSplicerExecStream::postViolation().

00058 {
00059     if (hasTarget()) {
00060         getErrorTarget().notifyError(
00061             name, level, message, address, capacity, index);
00062     }
00063 }

void ErrorSource::postError	(	ErrorLevel	level,
		const std::string &	message,
		const TupleDescriptor &	errorDesc,
		const TupleData &	errorTuple,
		int	index
	)			`[inherited]`

Posts an exception, such as a row exception.

See also:: ErrorTarget for a description of the parameters

Definition at line 65 of file ErrorSource.cpp.

References TupleAccessor::compute(), ErrorSource::errorAccessor, FixedBuffer, TupleAccessor::getByteCount(), TupleAccessor::getMaxByteCount(), ErrorSource::hasTarget(), TupleAccessor::marshal(), ErrorSource::pErrorBuf, and ErrorSource::postError().

00068 {
00069     if (!hasTarget()) {
00070         return;
00071     }
00072 
00073     if (!pErrorBuf) {
00074         errorAccessor.compute(errorDesc);
00075         uint cbMax = errorAccessor.getMaxByteCount();
00076         pErrorBuf.reset(new FixedBuffer[cbMax]);
00077     }
00078 
00079     uint cbTuple = errorAccessor.getByteCount(errorTuple);
00080     errorAccessor.marshal(errorTuple, pErrorBuf.get());
00081     postError(level, message, pErrorBuf.get(), cbTuple, index);
00082 }

bool ErrorSource::hasTarget ( ) const [inline, inherited]

Returns:: true iff an error target has been set

Definition at line 112 of file ErrorSource.h.

Referenced by ErrorSource::postError().

00113     {
00114         return pErrorTarget.get() ? true : false;
00115     }

ErrorTarget& ErrorSource::getErrorTarget ( ) const [inline, inherited]

Returns:: the ErrorTarget for this source

Definition at line 120 of file ErrorSource.h.

Referenced by ErrorSource::postError().

00121     {
00122         assert(hasTarget());
00123         return *(pErrorTarget.get());
00124     }

SharedErrorTarget ErrorSource::getSharedErrorTarget ( ) const [inline, inherited]

Returns:: the SharedErrorTarget for this source

Definition at line 129 of file ErrorSource.h.

00130     {
00131         return pErrorTarget;
00132     }

std::string ErrorSource::getErrorSourceName ( ) const [inline, inherited]

Gets the name of this source.

Useful to construct nested names for subcomponents that are also ErrorSources.

Returns:: the name

Definition at line 139 of file ErrorSource.h.

00140     {
00141         return name;
00142     }

void ErrorSource::setErrorSourceName ( std::string const & n ) [inline, inherited]

Sets the name of this source.

Useful to construct dynamic names for fine-grained filtering.

Definition at line 148 of file ErrorSource.h.

00149     {
00150         name = n;
00151     }

void ErrorSource::disableTarget ( ) [inherited]

Definition at line 84 of file ErrorSource.cpp.

References ErrorSource::pErrorTarget.

00085 {
00086     pErrorTarget.reset();
00087 }

Member Data Documentation

bool NestedLoopJoinExecStream::preProcessingDone [private]

True if pre-processing on third input completed.

Definition at line 81 of file NestedLoopJoinExecStream.h.

Referenced by open(), and preProcessRightInput().

std::vector<NestedLoopJoinKey> NestedLoopJoinExecStream::leftJoinKeys [private]

Dynamic parameters corresponding to the left join keys.

Definition at line 86 of file NestedLoopJoinExecStream.h.

Referenced by open(), prepare(), and processLeftInput().

bool CartesianJoinExecStream::leftOuter [protected, inherited]

Definition at line 61 of file CartesianJoinExecStream.h.

Referenced by CartesianJoinExecStream::execute(), and CartesianJoinExecStream::prepare().

bool CartesianJoinExecStream::rightInputEmpty [protected, inherited]

Definition at line 62 of file CartesianJoinExecStream.h.

Referenced by CartesianJoinExecStream::execute().

TupleData CartesianJoinExecStream::outputData [protected, inherited]

Definition at line 63 of file CartesianJoinExecStream.h.

Referenced by CartesianJoinExecStream::execute(), CartesianJoinExecStream::prepare(), and processLeftInput().

SharedExecStreamBufAccessor CartesianJoinExecStream::pLeftBufAccessor [protected, inherited]

Definition at line 64 of file CartesianJoinExecStream.h.

Referenced by CartesianJoinExecStream::execute(), open(), and CartesianJoinExecStream::prepare().

SharedExecStreamBufAccessor CartesianJoinExecStream::pRightBufAccessor [protected, inherited]

Definition at line 65 of file CartesianJoinExecStream.h.

Referenced by CartesianJoinExecStream::execute(), and CartesianJoinExecStream::prepare().

SharedExecStream CartesianJoinExecStream::pRightInput [protected, inherited]

Definition at line 66 of file CartesianJoinExecStream.h.

Referenced by CartesianJoinExecStream::execute(), and CartesianJoinExecStream::prepare().

uint CartesianJoinExecStream::nLeftAttributes [protected, inherited]

Definition at line 67 of file CartesianJoinExecStream.h.

Referenced by CartesianJoinExecStream::execute(), prepare(), and CartesianJoinExecStream::prepare().

std::vector<SharedExecStreamBufAccessor> ConfluenceExecStream::inAccessors [protected, inherited]

Definition at line 50 of file ConfluenceExecStream.h.

Referenced by checkNumInputs(), CartesianJoinExecStream::checkNumInputs(), LbmMinusExecStream::comparePrefixes(), LbmGeneratorExecStream::execute(), MergeExecStream::execute(), BarrierExecStream::execute(), LbmMinusExecStream::findMinInput(), LcsRowScanExecStream::initializeFiltersIfNeeded(), LcsRowScanExecStream::open(), LbmUnionExecStream::open(), LbmMinusExecStream::open(), LbmGeneratorExecStream::open(), LbmChopperExecStream::open(), LbmBitOpExecStream::open(), ConfluenceExecStream::open(), LcsRowScanExecStream::prepare(), LbmUnionExecStream::prepare(), LbmMinusExecStream::prepare(), LbmGeneratorExecStream::prepare(), LbmChopperExecStream::prepare(), LbmBitOpExecStream::prepare(), LhxJoinExecStream::prepare(), MergeExecStream::prepare(), CorrelationJoinExecStream::prepare(), ConfluenceExecStream::prepare(), CartesianJoinExecStream::prepare(), BarrierExecStream::prepare(), preProcessRightInput(), BarrierExecStream::processInputTuple(), LbmBitOpExecStream::producePendingOutput(), LbmMinusExecStream::restartSubtrahends(), LhxJoinExecStream::setHashInfo(), and ConfluenceExecStream::setInputBufAccessors().

SharedExecStreamBufAccessor SingleOutputExecStream::pOutAccessor [protected, inherited]

Definition at line 56 of file SingleOutputExecStream.h.

bool ExecStream::isOpen [protected, inherited]

Whether this stream is currently open.

Note that this is not quite the opposite of the inherited ClosableObject.needsClose, since a stream needs to be closed before destruction if it has been prepared but never opened.

Definition at line 61 of file ExecStream.h.

Referenced by ExecStream::closeImpl(), ExecStream::ExecStream(), and ExecStream::open().

ExecStreamGraph* ExecStream::pGraph [protected, inherited]

Dataflow graph containing this stream.

Note that we don't use a weak_ptr for this because it needs to be accessed frequently during execution, and the extra locking overhead would be frivolous.

Definition at line 68 of file ExecStream.h.

Referenced by ExecStream::checkAbort(), ExecStream::ExecStream(), CorrelationJoinExecStream::execute(), ExecStream::getGraph(), JavaTransformExecStream::open(), SingleInputExecStream::open(), MergeExecStream::open(), ExecStream::open(), CorrelationJoinExecStream::open(), ConfluenceExecStream::open(), ExecStream::prepare(), CartesianJoinExecStream::prepare(), and LbmMinusExecStream::restartSubtrahends().

ExecStreamId ExecStream::id [protected, inherited]

Identifier for this stream; local to its containing graph.

Definition at line 73 of file ExecStream.h.

Referenced by ExecStream::getStreamId().

std::string ExecStream::name [protected, inherited]

Name of stream, as known by optimizer.

Reimplemented from TraceSource.

Definition at line 78 of file ExecStream.h.

Referenced by ExecStream::ExecStream(), ExecStream::getName(), FlatFileExecStreamImpl::open(), and ExecStream::setName().

SharedDynamicParamManager ExecStream::pDynamicParamManager [protected, inherited]

The dynamic parameter manager available to this stream.

(Obtained at prepare() time. Keep a shared pointer in case the stream is reassigned to another graph for execution; cf ExecStreamGraph::mergeFrom())

Definition at line 85 of file ExecStream.h.

SharedLogicalTxn ExecStream::pTxn [protected, inherited]

The transaction embracing the stream.

Obtained at open() time; but not released at close() time, to allow TableWriters to replay a txn. Keep a shared pointer in case the stream is reassigned to another graph for execution; cf ExecStreamGraph::mergeFrom())

Definition at line 94 of file ExecStream.h.

Referenced by FtrsTableWriterExecStream::commitSavepoint(), FtrsTableWriterExecStream::createSavepoint(), FtrsTableWriterExecStream::open(), ExecStream::open(), and FtrsTableWriterExecStream::rollbackSavepoint().