LcsRowScanBaseExecStream Class Reference

Implements basic elements required to scan clusters in an exec stream. More...

#include <LcsRowScanBaseExecStream.h>

Inheritance diagram for LcsRowScanBaseExecStream:


Public Member Functions
	LcsRowScanBaseExecStream ()
virtual void	prepare (LcsRowScanBaseExecStreamParams const &params)
virtual void	open (bool restart)
	Opens this stream, acquiring any resources needed in order to be able to fetch data.
virtual void	getResourceRequirements (ExecStreamResourceQuantity &minQuantity, ExecStreamResourceQuantity &optQuantity)
virtual void	closeImpl ()
	Implements ClosableObject.
virtual void	prepare (ConfluenceExecStreamParams const &params)
virtual void	prepare (SingleOutputExecStreamParams const &params)
virtual void	prepare (ExecStreamParams const &params)
	Prepares this stream for execution.
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	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 ExecStreamResult	execute (ExecStreamQuantum const &quantum)=0
	Executes this stream.
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
void	syncColumns (SharedLcsClusterReader &pScan)
	Positions column readers based on new cluster reader position.
bool	readColVals (SharedLcsClusterReader &pScan, TupleDataWithBuffer &tupleData, uint colStart)
	Reads column values based on current position of cluster reader.
virtual void	buildOutputProj (TupleProjection &outputProj, LcsRowScanBaseExecStreamParams const &params)
	Builds outputProj from params.
Protected Attributes
VectorOfUint	projMap
	Projection map that maps columns read from cluster to their position in the output projection.
uint	nClusters
	Number of clusters to be scanned.
boost::scoped_array< SharedLcsClusterReader >	pClusters
	Array containing cluster readers.
TupleDescriptor	projDescriptor
	Tuple descriptor representing columns to be projected from scans.
std::vector< int >	nonClusterCols
	List of the non-cluster columns that need to be projected.
bool	allSpecial
	True in the special case where we are only reading special columns.
CircularBuffer< LcsRidRun >	ridRuns
	Circular buffer of rid runs.
std::vector< UnalignedAttributeAccessor >	attrAccessors
	Accessors used for loading actual column values.
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

Detailed Description

Implements basic elements required to scan clusters in an exec stream.

Definition at line 69 of file LcsRowScanBaseExecStream.h.

Constructor & Destructor Documentation

LcsRowScanBaseExecStream::LcsRowScanBaseExecStream ( ) [explicit]

Definition at line 28 of file LcsRowScanBaseExecStream.cpp.

References nClusters.

00029 {
00030     nClusters = 0;
00031 }

Member Function Documentation

void LcsRowScanBaseExecStream::syncColumns ( SharedLcsClusterReader & pScan ) [protected]

Positions column readers based on new cluster reader position.

Parameters:

pScan

cluster reader

Definition at line 172 of file LcsRowScanBaseExecStream.cpp.

Referenced by LbmGeneratorExecStream::advanceReader(), LcsRowScanExecStream::execute(), LbmGeneratorExecStream::execute(), and LbmGeneratorExecStream::generateMultiKeyBitmaps().

00173 {
00174     for (uint iCluCol = 0; iCluCol < pScan->nColsToRead; iCluCol++) {
00175         pScan->clusterCols[iCluCol].sync();
00176     }
00177 }

bool LcsRowScanBaseExecStream::readColVals	(	SharedLcsClusterReader &	pScan,
		TupleDataWithBuffer &	tupleData,
		uint	colStart
	)			`[protected]`

Reads column values based on current position of cluster reader.

Parameters:

	pScan	cluster reader
	tupleData	tupledata where data will be loaded
	colStart	starting column offset where first column will be loaded

Returns:: false if column filters failed; true otherwise

Definition at line 179 of file LcsRowScanBaseExecStream.cpp.

References allSpecial, attrAccessors, projDescriptor, and projMap.

Referenced by LcsRowScanExecStream::execute(), and LbmGeneratorExecStream::generateMultiKeyBitmaps().

00183 {
00184     if (!allSpecial) {
00185         for (uint iCluCol = 0; iCluCol < pScan->nColsToRead; iCluCol++) {
00186             // Get value of each column and load it to the appropriate
00187             // tuple datum entry
00188             PBuffer curValue = pScan->clusterCols[iCluCol].getCurrentValue();
00189             uint idx = projMap[colStart + iCluCol];
00190 
00191             attrAccessors[idx].loadValue(tupleData[idx], curValue);
00192             if (pScan->clusterCols[iCluCol].getFilters().hasResidualFilters) {
00193                 if (!pScan->clusterCols[iCluCol].applyFilters(
00194                     projDescriptor,
00195                     tupleData))
00196                 {
00197                     return false;
00198                 }
00199             }
00200         }
00201     }
00202     return true;
00203 }

void LcsRowScanBaseExecStream::buildOutputProj	(	TupleProjection &	outputProj,
		LcsRowScanBaseExecStreamParams const &	params
	)			`[protected, virtual]`

Builds outputProj from params.

Parameters:

	outputProj	the projection to be built
	params	the LcsRowScanBaseExecStreamParams

Reimplemented in LcsRowScanExecStream.

Definition at line 205 of file LcsRowScanBaseExecStream.cpp.

References LcsRowScanBaseExecStreamParams::outputProj.

Referenced by prepare().

00208 {
00209     /*
00210      * Copy the projection
00211      */
00212     for (uint i = 0;  i < params.outputProj.size(); i++) {
00213         outputProj.push_back(params.outputProj[i]);
00214     }
00215 }

void LcsRowScanBaseExecStream::prepare ( LcsRowScanBaseExecStreamParams const & params ) [virtual]

Definition at line 33 of file LcsRowScanBaseExecStream.cpp.

References allSpecial, buildOutputProj(), BTreeParams::keyProj, BTreeDescriptor::keyProjection, LCS_RID_COLUMN_ID, LcsRowScanBaseExecStreamParams::lcsClusterScanDefs, nClusters, nonClusterCols, LcsRowScanBaseExecStreamParams::outputProj, BTreeParams::pageOwnerId, BTreeDescriptor::pageOwnerId, ExecStreamParams::pCacheAccessor, SegmentAccessor::pCacheAccessor, pClusters, SingleOutputExecStream::pOutAccessor, ConfluenceExecStream::prepare(), projDescriptor, projMap, BTreeParams::pSegment, SegmentAccessor::pSegment, ridRuns, BTreeParams::rootPageId, BTreeDescriptor::rootPageId, BTreeDescriptor::segmentAccessor, BTreeParams::segmentId, BTreeDescriptor::segmentId, BTreeParams::tupleDesc, and BTreeDescriptor::tupleDescriptor.

Referenced by LcsRowScanExecStream::prepare(), and LbmGeneratorExecStream::prepare().

00035 {
00036     ConfluenceExecStream::prepare(params);
00037 
00038     // Copy cluster definition parameters and setup btree readers for each
00039     // cluster.  Also, setup the full output tuple based on the ordered
00040     // list of cluster descriptors.
00041 
00042     nClusters = params.lcsClusterScanDefs.size();
00043     pClusters.reset(new SharedLcsClusterReader[nClusters]);
00044 
00045     uint clusterStart = 0;
00046     uint projCount = 0;
00047     TupleDescriptor allClusterTupleDesc;
00048     TupleProjection newProj, outputProj;
00049 
00050     buildOutputProj(outputProj, params);
00051 
00052     newProj.resize(outputProj.size());
00053 
00054     // if we're projecting non-cluster columns, keep track of them separately;
00055     TupleDescriptor outputTupleDesc = pOutAccessor->getTupleDesc();
00056     for (uint i = 0; i < outputProj.size(); i++) {
00057         if (outputProj[i] == LCS_RID_COLUMN_ID) {
00058             newProj[i] = projCount++;
00059             allClusterTupleDesc.push_back(outputTupleDesc[i]);
00060             nonClusterCols.push_back(params.outputProj[i]);
00061         }
00062     }
00063 
00064     allSpecial = (nonClusterCols.size() == newProj.size());
00065 
00066     for (uint i = 0; i < nClusters; i++) {
00067         SharedLcsClusterReader &pClu = pClusters[i];
00068 
00069         BTreeExecStreamParams const &bTreeParams = params.lcsClusterScanDefs[i];
00070 
00071         BTreeDescriptor treeDescriptor;
00072         treeDescriptor.segmentAccessor.pSegment = bTreeParams.pSegment;
00073         treeDescriptor.segmentAccessor.pCacheAccessor =
00074             bTreeParams.pCacheAccessor;
00075         treeDescriptor.tupleDescriptor = bTreeParams.tupleDesc;
00076         treeDescriptor.keyProjection = bTreeParams.keyProj;
00077         treeDescriptor.rootPageId = bTreeParams.rootPageId;
00078         treeDescriptor.segmentId = bTreeParams.segmentId;
00079         treeDescriptor.pageOwnerId = bTreeParams.pageOwnerId;
00080 
00081         pClu =
00082             SharedLcsClusterReader(
00083                 new LcsClusterReader(treeDescriptor, &ridRuns));
00084 
00085         // setup the cluster and column readers to only read the columns
00086         // that are going to be projected
00087         uint clusterEnd = clusterStart +
00088             params.lcsClusterScanDefs[i].clusterTupleDesc.size() - 1;
00089 
00090         // create a vector of the columns that are projected from
00091         // this cluster and recompute the projection list
00092         // based on the individual cluster projections
00093         TupleProjection clusterProj;
00094         for (uint j = 0; j < newProj.size(); j++) {
00095             if (outputProj[j] >= clusterStart &&
00096                 outputProj[j] <= clusterEnd)
00097             {
00098                 clusterProj.push_back(outputProj[j] - clusterStart);
00099                 newProj[j] = projCount++;
00100             }
00101         }
00102         clusterStart = clusterEnd + 1;
00103 
00104         // need to select at least one column from cluster, except in the
00105         // cases where we're only selecting special columns or when there
00106         // are filter columns; in the former case, we'll just arbitrarily
00107         // read the first column, but not actually project it
00108         if (allSpecial) {
00109            clusterProj.push_back(0);
00110         }
00111         pClu->initColumnReaders(
00112             params.lcsClusterScanDefs[i].clusterTupleDesc.size(),
00113             clusterProj);
00114         if (!allSpecial) {
00115             for (uint j = 0; j < pClu->nColsToRead; j++) {
00116                 allClusterTupleDesc.push_back(
00117                     params.lcsClusterScanDefs[i].
00118                         clusterTupleDesc[clusterProj[j]]);
00119             }
00120         }
00121     }
00122 
00123     // setup projected tuple descriptor, by reshuffling allClusterTupleDesc
00124     // built above, into the correct projection order
00125 
00126     for (uint i = 0; i < newProj.size(); i++) {
00127         projDescriptor.push_back(allClusterTupleDesc[newProj[i]]);
00128     }
00129 
00130     // create a projection map to map cluster data read to the output
00131     // projection
00132     projMap.resize(newProj.size());
00133     for (uint i = 0; i < projMap.size(); i++) {
00134         for (uint j = 0; j < newProj.size(); j++) {
00135             if (newProj[j] == i) {
00136                 projMap[i] = j;
00137             }
00138         }
00139     }
00140 }

void LcsRowScanBaseExecStream::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 ConfluenceExecStream.

Reimplemented in LbmGeneratorExecStream, and LcsRowScanExecStream.

Definition at line 142 of file LcsRowScanBaseExecStream.cpp.

References nClusters, ConfluenceExecStream::open(), and pClusters.

Referenced by LcsRowScanExecStream::open(), and LbmGeneratorExecStream::open().

00143 {
00144     ConfluenceExecStream::open(restart);
00145     for (uint i = 0; i < nClusters; i++) {
00146         pClusters[i]->open();
00147     }
00148 }

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

Reimplemented from ExecStream.

Reimplemented in LcsRowScanExecStream.

Definition at line 150 of file LcsRowScanBaseExecStream.cpp.

References ExecStream::getResourceRequirements(), ExecStreamResourceQuantity::nCachePages, and nClusters.

Referenced by LcsRowScanExecStream::getResourceRequirements(), and LbmGeneratorExecStream::getResourceRequirements().

00153 {
00154     ConfluenceExecStream::getResourceRequirements(minQuantity, optQuantity);
00155 
00156     // 2 pages per cluster (not taking into account pre-fetches yet)
00157     // - 1 for cluster page
00158     // - 1 for btree page
00159     minQuantity.nCachePages += (nClusters * 2);
00160 
00161     optQuantity = minQuantity;
00162 }

void LcsRowScanBaseExecStream::closeImpl ( ) [virtual]

Implements ClosableObject.

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

Reimplemented from ExecStream.

Reimplemented in LbmGeneratorExecStream, and LcsRowScanExecStream.

Definition at line 164 of file LcsRowScanBaseExecStream.cpp.

References ClosableObject::close(), ExecStream::closeImpl(), nClusters, and pClusters.

Referenced by LcsRowScanExecStream::closeImpl(), and LbmGeneratorExecStream::closeImpl().

00165 {
00166     ConfluenceExecStream::closeImpl();
00167     for (uint i = 0; i < nClusters; i++) {
00168         pClusters[i]->close();
00169     }
00170 }

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 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 }

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 }

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::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 }

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

Executes this stream.

Parameters:

quantum

governs the maximum amount of execution to perform

Returns:: code indicating reason execution ceased

Implemented in CalcExecStream, BarrierExecStream, BernoulliSamplingExecStream, CartesianJoinExecStream, CollectExecStream, CopyExecStream, CorrelationJoinExecStream, DoubleBufferExecStream, MergeExecStream, MockConsumerExecStream, MockProducerExecStream, MockResourceExecStream, ReshapeExecStream, ScratchBufferExecStream, SegBufferExecStream, SegBufferReaderExecStream, SegBufferWriterExecStream, SortedAggExecStream, SplitterExecStream, UncollectExecStream, ValuesExecStream, JavaSinkExecStream, JavaTransformExecStream, FlatFileExecStreamImpl, BTreeInsertExecStream, BTreePrefetchSearchExecStream, BTreeScanExecStream, BTreeSearchExecStream, BTreeSearchUniqueExecStream, BTreeSortExecStream, FtrsTableWriterExecStream, LhxAggExecStream, LhxJoinExecStream, LbmChopperExecStream, LbmGeneratorExecStream, LbmIntersectExecStream, LbmMinusExecStream, LbmNormalizerExecStream, LbmSplicerExecStream, LbmUnionExecStream, LcsClusterAppendExecStream, LcsRowScanExecStream, and ExternalSortExecStreamImpl.

Referenced by ExecStreamScheduler::executeStream().

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(), 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

VectorOfUint LcsRowScanBaseExecStream::projMap [protected]

Projection map that maps columns read from cluster to their position in the output projection.

Definition at line 77 of file LcsRowScanBaseExecStream.h.

Referenced by LcsRowScanExecStream::execute(), LbmGeneratorExecStream::generateSingletons(), prepare(), LcsRowScanExecStream::prepareResidualFilters(), and readColVals().

uint LcsRowScanBaseExecStream::nClusters [protected]

Number of clusters to be scanned.

Definition at line 82 of file LcsRowScanBaseExecStream.h.

Referenced by closeImpl(), LcsRowScanExecStream::execute(), LbmGeneratorExecStream::execute(), LbmGeneratorExecStream::generateMultiKeyBitmaps(), getResourceRequirements(), LcsRowScanBaseExecStream(), open(), LbmGeneratorExecStream::open(), prepare(), LbmGeneratorExecStream::prepare(), and LcsRowScanExecStream::prepareResidualFilters().

boost::scoped_array<SharedLcsClusterReader> LcsRowScanBaseExecStream::pClusters [protected]

Array containing cluster readers.

Definition at line 87 of file LcsRowScanBaseExecStream.h.

Referenced by closeImpl(), LcsRowScanExecStream::execute(), LbmGeneratorExecStream::execute(), LbmGeneratorExecStream::generateBitmaps(), LbmGeneratorExecStream::generateMultiKeyBitmaps(), LbmGeneratorExecStream::generateSingleKeyBitmaps(), LbmGeneratorExecStream::generateSingletons(), open(), LbmGeneratorExecStream::open(), prepare(), and LcsRowScanExecStream::prepareResidualFilters().

TupleDescriptor LcsRowScanBaseExecStream::projDescriptor [protected]

Tuple descriptor representing columns to be projected from scans.

Definition at line 92 of file LcsRowScanBaseExecStream.h.

Referenced by LcsRowScanExecStream::prepare(), prepare(), LcsRowScanExecStream::prepareResidualFilters(), and readColVals().

std::vector<int> LcsRowScanBaseExecStream::nonClusterCols [protected]

List of the non-cluster columns that need to be projected.

Definition at line 97 of file LcsRowScanBaseExecStream.h.

Referenced by LcsRowScanExecStream::execute(), prepare(), and LcsRowScanExecStream::prepareResidualFilters().

bool LcsRowScanBaseExecStream::allSpecial [protected]

True in the special case where we are only reading special columns.

I.e., we don't actually have to read the underlying cluster data.

Definition at line 103 of file LcsRowScanBaseExecStream.h.

Referenced by prepare(), and readColVals().

CircularBuffer<LcsRidRun> LcsRowScanBaseExecStream::ridRuns [protected]

Circular buffer of rid runs.

Definition at line 108 of file LcsRowScanBaseExecStream.h.

Referenced by LcsRowScanExecStream::execute(), LbmGeneratorExecStream::execute(), LcsRowScanExecStream::fillRidRunBuffer(), LcsRowScanExecStream::LcsRowScanExecStream(), LcsRowScanExecStream::open(), LbmGeneratorExecStream::open(), prepare(), and LbmGeneratorExecStream::prepare().

std::vector<UnalignedAttributeAccessor> LcsRowScanBaseExecStream::attrAccessors [protected]

Accessors used for loading actual column values.

Definition at line 120 of file LcsRowScanBaseExecStream.h.

Referenced by LbmGeneratorExecStream::generateBitmaps(), LbmGeneratorExecStream::generateSingletons(), LcsRowScanExecStream::prepare(), LbmGeneratorExecStream::prepare(), and readColVals().

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

Definition at line 50 of file ConfluenceExecStream.h.

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().