LcsRowScanExecStream Class Reference

Given a stream of RIDs, performs a table scan for those RIDs using the appropriate clustered indexes defined on the table. More...

#include <LcsRowScanExecStream.h>

Inheritance diagram for LcsRowScanExecStream:

List of all members.

Public Member Functions
	LcsRowScanExecStream ()
virtual void	prepare (LcsRowScanExecStreamParams const &params)
virtual void	open (bool restart)
	Opens this stream, acquiring any resources needed in order to be able to fetch data.
virtual ExecStreamResult	execute (ExecStreamQuantum const &quantum)
	Executes this stream.
virtual void	getResourceRequirements (ExecStreamResourceQuantity &minQuantity, ExecStreamResourceQuantity &optQuantity)
virtual void	closeImpl ()
	Implements ClosableObject.
virtual void	prepare (LcsRowScanBaseExecStreamParams 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 void	getResourceRequirements (ExecStreamResourceQuantity &minQuantity, ExecStreamResourceQuantity &optQuantity, ExecStreamResourceSettingType &optType)
	Determines resource requirements for 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	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
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.
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
Private Member Functions
virtual void	buildOutputProj (TupleProjection &outputProj, LcsRowScanBaseExecStreamParams const &params)
	Builds outputProj from params.
bool	initializeFiltersIfNeeded ()
	initializes the filter data structures
void	prepareResidualFilters (LcsRowScanExecStreamParams const &params)
	initializes the filter data structures during prepare time
void	initializeSystemSampling ()
	Initializes the system sampling data structures during open time.
ExecStreamResult	fillRidRunBuffer ()
	Populates the circular rid run buffer.
Private Attributes
TupleDataWithBuffer	outputTupleData
	Tuple data for all columns read from all clusters, including filter columns.
uint	iFilterToInitialize
	This variable is used to control the initialization of residual filters.
TupleData	projOutputTupleData
TupleProjection	outputProj
TupleData	ridTupleData
	Tuple data for input stream.
LbmRidReader	ridReader
	Rid reader.
RecordNum	nRidsRead
	Number of rids read.
LcsRid	inputRid
	Current rid read from the input stream.
LcsRid	nextRid
	Next rid that needs to be fetched.
bool	readDeletedRid
	True if need to read a new deleted rid from the input stream.
bool	deletedRidEos
	True if reached EOS on deleted rid input stream.
LcsRid	deletedRid
	Current deleted rid.
bool	tupleFound
	true if tuple has been read and not yet produced
bool	isFullScan
	true if executing full table scan
bool	hasExtraFilter
	true if there's extra range list filter(as the last input)
bool	producePending
	true if produceTuple pending
boost::scoped_array< LcsResidualColumnFilters * >	filters
	The local filter data structure.
int32_t	nFilters
	The number of residual column filters configured.
TableSamplingMode	samplingMode
	One of SAMPLING_OFF, SAMPLING_BERNOULLI or SAMPLING_SYSTEM.
float	samplingRate
	the sampling rate (0.0 to 1.0)
bool	isSamplingRepeatable
	true if the sample should be repeatable
int32_t	repeatableSeed
	seed for repeatable sampling
int32_t	samplingClumps
	number of clumps for system sampling
uint64_t	clumpSize
	size of each sampling clump
uint64_t	clumpDistance
	distance (in rows) between each clump
uint64_t	clumpPos
	position (0 to clumpSize) in current clump
uint64_t	clumpSkipPos
	position (clumpDistance to 0) in between clumps
uint	numClumps
	The number of clumps that need to be built.
uint	numClumpsBuilt
	Running counter of the number of clumps built.
boost::scoped_ptr< BernoulliRng >	samplingRng
	RNG for Bernoulli sampling.
int64_t	rowCount
	Number of rows in the table.
bool	ridRunsBuilt
	True if completed building rid runs.
LcsRidRun	currRidRun
	Current rid run being constructed.
CircularBufferIter< LcsRidRun >	ridRunIter
	Iterator over the circular buffer containing rid runs.

---

Detailed Description

Given a stream of RIDs, performs a table scan for those RIDs using the appropriate clustered indexes defined on the table.

The stream returns a projected subset of columns from the table

Definition at line 110 of file LcsRowScanExecStream.h.

---

Constructor & Destructor Documentation

LcsRowScanExecStream::LcsRowScanExecStream

(

)

Definition at line 33 of file LcsRowScanExecStream.cpp.

References CircularBuffer< T >::resize(), and LcsRowScanBaseExecStream::ridRuns.

:
    LcsRowScanBaseExecStream(),
    ridRunIter(&ridRuns)
{
    ridRuns.resize(4000);
}

---

Member Function Documentation

void LcsRowScanExecStream::buildOutputProj	(	TupleProjection &	outputProj,
		LcsRowScanBaseExecStreamParams const &	params
	)			[private, virtual]

Builds outputProj from params.

Parameters:

	outputProj	the projection to be built
	params	the LcsRowScanBaseExecStreamParams

Reimplemented from LcsRowScanBaseExecStream.

Definition at line 681 of file LcsRowScanExecStream.cpp.

References outputProj, LcsRowScanBaseExecStreamParams::outputProj, and LcsRowScanExecStreamParams::residualFilterCols.

{
    LcsRowScanExecStreamParams const &rowScanParams =
        dynamic_cast<const LcsRowScanExecStreamParams&>(params);

    /*
     * Build a projection that contains filter columns
     */
    for (uint i = 0; i < rowScanParams.outputProj.size(); i++) {
        outputProj.push_back(rowScanParams.outputProj[i]);
    }
    for (uint i = 0; i < rowScanParams.residualFilterCols.size(); i++) {
        uint j;
        for (j = 0; j < rowScanParams.outputProj.size(); j++) {
            if (rowScanParams.outputProj[j] ==
                rowScanParams.residualFilterCols[i])
            {
                break;
            }
        }

        if (j >= rowScanParams.outputProj.size()) {
            outputProj.push_back(rowScanParams.residualFilterCols[i]);
        }
    }
}

bool LcsRowScanExecStream::initializeFiltersIfNeeded

(

)

[private]

initializes the filter data structures

Returns:

false iff input under flows.

Definition at line 256 of file LcsRowScanExecStream.cpp.

References EXECBUF_EOS, filters, FixedBuffer, TupleAccessor::getCurrentByteCount(), TupleAccessor::getCurrentTupleBuf(), iFilterToInitialize, ConfluenceExecStream::inAccessors, nFilters, TupleAccessor::setCurrentTupleBuf(), and TupleAccessor::unmarshal().

Referenced by execute().

{
    /*
     * initialize the filters local data
     */
    for (; iFilterToInitialize < nFilters; iFilterToInitialize++) {
        SharedExecStreamBufAccessor &pInAccessor =
            inAccessors[iFilterToInitialize + 1];
        TupleAccessor &inputAccessor =
            pInAccessor->getConsumptionTupleAccessor();

        if (pInAccessor->getState() != EXECBUF_EOS) {
            PLcsResidualColumnFilters filter = filters[iFilterToInitialize];

            while (pInAccessor->demandData()) {
                SharedLcsResidualFilter filterData(new LcsResidualFilter);

                pInAccessor->accessConsumptionTuple();

                /*
                 * Build lower and upper bound data
                 */
                filterData->boundData.compute(pInAccessor->getTupleDesc());
                filterData->boundBuf.reset(
                    new FixedBuffer[inputAccessor.getCurrentByteCount()]);

                memcpy(
                    filterData->boundBuf.get(),
                    pInAccessor->getConsumptionStart(),
                    inputAccessor.getCurrentByteCount());

                /*
                 * inputAccessor is used to unmarshal into boundData.
                 * in order to do this, its current buffer is set to
                 * boundBuf and restored.
                 */
                PConstBuffer tmpBuf;
                tmpBuf = inputAccessor.getCurrentTupleBuf();
                inputAccessor.setCurrentTupleBuf(filterData->boundBuf.get());
                inputAccessor.unmarshal(filterData->boundData);
                inputAccessor.setCurrentTupleBuf(tmpBuf);

                /*
                 * record directives.
                 */
                filterData->lowerBoundDirective =
                    SearchEndpoint(*filterData->boundData[0].pData);
                filterData->upperBoundDirective =
                    SearchEndpoint(*filterData->boundData[2].pData);

                filter->filterData.push_back(filterData);

                pInAccessor->consumeTuple();
            }

            if (pInAccessor->getState() != EXECBUF_EOS) {
                return false;
            }
        }
        filters[iFilterToInitialize]->filterDataInitialized = true;
    }
    return true;
}

void LcsRowScanExecStream::prepareResidualFilters

(

LcsRowScanExecStreamParams const &

params

)

[private]

initializes the filter data structures during prepare time

Parameters:

params

the LcsRowScanExecStreamParams

Definition at line 41 of file LcsRowScanExecStream.cpp.

References LcsResidualColumnFilters::attrAccessor, UnalignedAttributeAccessor::compute(), TupleDataWithBuffer::computeAndAllocate(), filters, LcsResidualColumnFilters::hasResidualFilters, LcsResidualColumnFilters::inputKeyDesc, LcsRowScanBaseExecStreamParams::lcsClusterScanDefs, LcsResidualColumnFilters::lowerBoundProj, LcsRowScanBaseExecStream::nClusters, nFilters, LcsRowScanBaseExecStream::nonClusterCols, LcsRowScanBaseExecStreamParams::outputProj, LcsRowScanBaseExecStream::pClusters, LcsRowScanBaseExecStream::projDescriptor, TupleDescriptor::projectFrom(), LcsRowScanBaseExecStream::projMap, LcsResidualColumnFilters::readerKeyData, LcsResidualColumnFilters::readerKeyProj, LcsRowScanExecStreamParams::residualFilterCols, and LcsResidualColumnFilters::upperBoundProj.

Referenced by prepare().

{
    nFilters = params.residualFilterCols.size();

    /*
     * compute the outputTupleData position of filter columns
     */
    VectorOfUint valueCols;
    uint j, k = 0;
    for (uint i = 0;  i < nFilters; i++) {
        for (j = 0; j < params.outputProj.size(); j++) {
            if (params.outputProj[j] == params.residualFilterCols[i]) {
                valueCols.push_back(j);
                break;
            }
        }

        if (j >= params.outputProj.size()) {
            valueCols.push_back(params.outputProj.size() +  k);
            k++;
        }
    }

    /*
     * compute the cluster id and cluster position
     */
    uint valueClus;
    uint clusterPos;
    uint clusterStart = 0;
    uint realClusterStart = 0;

    filters.reset(new PLcsResidualColumnFilters[nFilters]);

    for (uint i = 0; i < nClusters; i++) {
        uint clusterEnd = clusterStart +
            params.lcsClusterScanDefs[i].clusterTupleDesc.size() - 1;

        for (uint j = 0; j < nFilters; j++) {
            if (params.residualFilterCols[j] >= clusterStart &&
                params.residualFilterCols[j] <= clusterEnd)
            {
                valueClus = i;

                /*
                 * find the position within the cluster
                 */
                for (uint k = 0; k < projMap.size(); k++) {
                    if (projMap[k] == valueCols[j]) {
                        clusterPos = k - realClusterStart -
                            nonClusterCols.size();

                        LcsResidualColumnFilters &filter =
                            pClusters[valueClus]->
                            clusterCols[clusterPos].
                            getFilters();

                        filters[j] = &filter;

                        filter.hasResidualFilters = true;

                        filter.readerKeyProj.push_back(valueCols[j]);
                        filter.inputKeyDesc.projectFrom(
                            projDescriptor,
                            filter.readerKeyProj);
                        filter.attrAccessor.compute(
                            filter.inputKeyDesc[0]);

                        filter.lowerBoundProj.push_back(1);
                        filter.upperBoundProj.push_back(3);
                        filter.readerKeyData.computeAndAllocate(
                            filter.inputKeyDesc);

                        break;
                    }
                }
                // Continue with the same cluster for more filters
            }
        }
        // Look for filters in the next cluster; modify cluster boundaries
        clusterStart = clusterEnd + 1;
        realClusterStart += pClusters[i]->nColsToRead;
    }
}

void LcsRowScanExecStream::initializeSystemSampling

(

)

[private]

Initializes the system sampling data structures during open time.

Definition at line 321 of file LcsRowScanExecStream.cpp.

References clumpDistance, clumpPos, clumpSize, clumpSkipPos, numClumps, rowCount, samplingClumps, samplingRate, and TRACE_FINE.

Referenced by open().

{
    clumpPos = 0;
    clumpSkipPos = 0;

    FENNEL_TRACE(TRACE_FINE, "rowCount = " << rowCount);
    FENNEL_TRACE(
        TRACE_FINE, "samplingRate = " << static_cast<double>(samplingRate));

    if (rowCount <= 0) {
        // Handle empty table or non-sense input.
        clumpSize = 1;
        clumpDistance = 0;
        numClumps = 0;
        return;
    }

    // Manipulate this value in a separate member field so we don't
    // mistakenly modify our stored copy of the parameter.
    numClumps = samplingClumps;

    // Compute clump size and distance
    int64_t sampleSize =
        static_cast<uint64_t>(
            round(
                static_cast<double>(rowCount) *
                static_cast<double>(samplingRate)));
    if (sampleSize < numClumps) {
        // Read at least as many rows as there are clumps, even if sample rate
        // is very small.
        sampleSize = numClumps;
    }

    if (sampleSize > rowCount) {
        // samplingRate should be < 1.0, but handle the case where it isn't,
        // or where there are fewer rows than clumps.
        sampleSize = rowCount;
        numClumps = 1;
    }

    FENNEL_TRACE(TRACE_FINE, "sampleSize = " << sampleSize);

    clumpSize =
        static_cast<uint64_t>(
            round(
                static_cast<double>(sampleSize) /
                static_cast<double>(numClumps)));
    assert(sampleSize >= clumpSize);
    assert(clumpSize >= 1);

    FENNEL_TRACE(TRACE_FINE, "clumpSize = " << clumpSize);

    if (numClumps > 1) {
        // Arrange for the last clump to end at the end of the table.
        clumpDistance =
            static_cast<uint64_t>(
                round(
                    static_cast<double>(rowCount - sampleSize) /
                    static_cast<double>(numClumps - 1)));

        // Rounding can cause us to push the final clump past the end of the
        // table.  Avoid this when possible.
        uint64_t rowsRequired =
            (clumpSize + clumpDistance) * (numClumps - 1) + clumpSize;
        if (rowsRequired > rowCount && clumpDistance > 0) {
            clumpDistance--;
        }
    } else {
        // The entire sample will come from the beginning of the table.
        clumpDistance = (rowCount - sampleSize);
    }

    FENNEL_TRACE(TRACE_FINE, "clumpDistance = " << clumpDistance);
}

ExecStreamResult LcsRowScanExecStream::fillRidRunBuffer

(

)

[private]

Populates the circular rid run buffer.

Returns:

EXECRC_YIELD if buffer successfully populated

Definition at line 520 of file LcsRowScanExecStream.cpp.

References clumpDistance, clumpPos, clumpSize, clumpSkipPos, currRidRun, deletedRid, deletedRidEos, EXECRC_EOS, EXECRC_YIELD, inputRid, isFullScan, MAXU, LcsRidRun::nRids, numClumps, numClumpsBuilt, opaqueToInt(), CircularBuffer< T >::push_back(), readDeletedRid, LbmRidReaderBase::readRidAndAdvance(), ridReader, LcsRowScanBaseExecStream::ridRuns, ridRunsBuilt, rowCount, SAMPLING_OFF, SAMPLING_SYSTEM, samplingMode, samplingRng, CircularBuffer< T >::setReadOnly(), CircularBuffer< T >::spaceAvailable(), and LcsRidRun::startRid.

Referenced by execute().

{
    ExecStreamResult rc;
    RecordNum nRows;

    do {
        if (!isFullScan) {
            rc = ridReader.readRidAndAdvance(inputRid);
            if (rc == EXECRC_EOS) {
                ridRunsBuilt = true;
                break;
            }
            if (rc != EXECRC_YIELD) {
                return rc;
            }
            nRows = 1;

        } else {
            if (!deletedRidEos && readDeletedRid) {
                rc = ridReader.readRidAndAdvance(deletedRid);
                if (rc == EXECRC_EOS) {
                    deletedRidEos = true;
                    if (samplingMode == SAMPLING_OFF) {
                        ridRunsBuilt = true;
                    } else if (samplingMode == SAMPLING_SYSTEM &&
                        numClumps == 0)
                    {
                        ridRunsBuilt = true;
                        break;
                    }
                } else if (rc != EXECRC_YIELD) {
                    return rc;
                } else {
                    readDeletedRid = false;
                }
            }
            // skip over deleted rids
            if (!deletedRidEos && inputRid == deletedRid) {
                inputRid++;
                readDeletedRid = true;
                continue;
            } else {
                if (deletedRidEos) {
                    nRows = MAXU;
                } else {
                    nRows = opaqueToInt(deletedRid - inputRid);
                }
            }
        }

        if (samplingMode != SAMPLING_OFF) {
            if (samplingMode == SAMPLING_SYSTEM) {
                if (clumpSkipPos > 0) {
                    // We need to skip clumpSkipPos RIDs, taking into
                    // account deleted RIDs.  If all deleted RIDs have been
                    // processed (a), we can just skip forward to the next
                    // clump.  If we know the next deleted RID, skip to the
                    // next clump if we can (b), else skip to the deleted
                    // RID (c).  Processing will return here to handle the
                    // remaining clumpSkipPos rows when we reach the next
                    // live RID.  If we don't know the next deleted RID
                    // (d), skip the current live RID, let the deleted RID
                    // processing occur above and then processing will
                    // return here to deal with the remaining clumpSkipPos
                    // rows.
                    if (deletedRidEos) {
                        // (a)
                        inputRid += clumpSkipPos;
                        clumpSkipPos = 0;
                    } else if (!readDeletedRid) {
                        if (deletedRid > inputRid + clumpSkipPos) {
                            // (b)
                            inputRid += clumpSkipPos;
                            clumpSkipPos = 0;
                            nRows = opaqueToInt(deletedRid - inputRid);
                        } else {
                            // (c)
                            clumpSkipPos -= opaqueToInt(deletedRid - inputRid);
                            inputRid = deletedRid;
                            continue;
                        }
                    } else {
                        // (d)
                        clumpSkipPos--;
                        inputRid++;
                        continue;
                    }
                }

                if (nRows >= clumpSize - clumpPos) {
                    // Scale back the size of the rid run based on the
                    // clump size
                    nRows = clumpSize - clumpPos;
                    clumpPos = 0;
                    clumpSkipPos = clumpDistance;
                    if (++numClumpsBuilt == numClumps) {
                        ridRunsBuilt = true;
                    }
                } else {
                    // We only have enough rids for a partial clump
                    clumpPos += nRows;
                }
            } else {
                // Bernoulli sampling
                if (opaqueToInt(inputRid) >= opaqueToInt(rowCount)) {
                    ridRunsBuilt = true;
                    break;
                }
                if (!samplingRng->nextValue()) {
                    inputRid++;
                    continue;
                }
                nRows = 1;
            }
        }

        if (currRidRun.startRid == LcsRid(MAXU)) {
            currRidRun.startRid = inputRid;
            currRidRun.nRids = nRows;
        } else if (currRidRun.startRid + currRidRun.nRids == inputRid) {
            // If the next set of rids is contiguous with the previous,
            // continue adding on to the current run
            if (nRows == RecordNum(MAXU)) {
                currRidRun.nRids = MAXU;
            } else {
                currRidRun.nRids += nRows;
            }
        } else {
            // Otherwise, end the current one
            ridRuns.push_back(currRidRun);

            // And start a new one
            currRidRun.startRid = inputRid;
            currRidRun.nRids = nRows;
        }

        if (isFullScan) {
            inputRid += nRows;
        }
    } while (ridRuns.spaceAvailable() && !ridRunsBuilt);

    // Write out the last run
    if (ridRunsBuilt && currRidRun.startRid != LcsRid(MAXU)) {
        ridRuns.push_back(currRidRun);
    }

    if (ridRunsBuilt) {
        ridRuns.setReadOnly();
    }
    return EXECRC_YIELD;
}

void LcsRowScanExecStream::prepare

(

LcsRowScanExecStreamParams const &

params

)

[virtual]

Definition at line 126 of file LcsRowScanExecStream.cpp.

References LcsRowScanBaseExecStream::attrAccessors, TupleData::compute(), TupleDataWithBuffer::computeAndAllocate(), LcsRowScanExecStreamParams::hasExtraFilter, hasExtraFilter, ConfluenceExecStream::inAccessors, LcsRowScanExecStreamParams::isFullScan, isFullScan, isSamplingRepeatable, nFilters, outputProj, LcsRowScanBaseExecStreamParams::outputProj, outputTupleData, SingleOutputExecStream::pOutAccessor, LcsRowScanBaseExecStream::prepare(), prepareResidualFilters(), LcsRowScanBaseExecStream::projDescriptor, projOutputTupleData, repeatableSeed, ridTupleData, rowCount, SAMPLING_BERNOULLI, SAMPLING_OFF, LcsRowScanExecStreamParams::samplingClumps, samplingClumps, LcsRowScanExecStreamParams::samplingIsRepeatable, LcsRowScanExecStreamParams::samplingMode, samplingMode, LcsRowScanExecStreamParams::samplingRate, samplingRate, LcsRowScanExecStreamParams::samplingRepeatableSeed, samplingRng, LcsRowScanExecStreamParams::samplingRowCount, and STANDARD_TYPE_RECORDNUM.

{
    LcsRowScanBaseExecStream::prepare(params);

    isFullScan = params.isFullScan;
    hasExtraFilter = params.hasExtraFilter;

    // Set up rid bitmap input stream
    ridTupleData.compute(inAccessors[0]->getTupleDesc());

    // validate input stream parameters
    TupleDescriptor inputDesc = inAccessors[0]->getTupleDesc();
    assert(inputDesc.size() == 3);
    StandardTypeDescriptorFactory stdTypeFactory;
    TupleAttributeDescriptor expectedRidDesc(
        stdTypeFactory.newDataType(STANDARD_TYPE_RECORDNUM));
    assert(inputDesc[0] == expectedRidDesc);

    assert(hasExtraFilter == (inAccessors.size() > 1));

    if (hasExtraFilter) {
        prepareResidualFilters(params);
    } else {
        nFilters = 0;
    }

    /*
     * projDescriptor now also includes filter columns
     */
    for (uint i = 0; i < params.outputProj.size(); i++) {
        outputProj.push_back(i);
    }

    pOutAccessor->setTupleShape(pOutAccessor->getTupleDesc());
    outputTupleData.computeAndAllocate(projDescriptor);

    /*
     * build the real output accessor
     * it will be used to unmarshal data into the
     * real output row: projOutputTuple.
     */
    projOutputTupleData.compute(pOutAccessor->getTupleDesc());

    attrAccessors.resize(projDescriptor.size());
    for (uint i = 0; i < projDescriptor.size(); ++i) {
        attrAccessors[i].compute(projDescriptor[i]);
    }

    /* configure sampling */
    samplingMode = params.samplingMode;

    if (samplingMode != SAMPLING_OFF) {
        samplingRate = params.samplingRate;
        rowCount = params.samplingRowCount;

        if (samplingMode == SAMPLING_BERNOULLI) {
            isSamplingRepeatable = params.samplingIsRepeatable;
            repeatableSeed = params.samplingRepeatableSeed;
            samplingClumps = -1;

            samplingRng.reset(new BernoulliRng(samplingRate));
        } else {
            assert(isFullScan);

            samplingClumps = params.samplingClumps;
            assert(samplingClumps > 0);

            isSamplingRepeatable = false;
        }
    }
}

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

Definition at line 198 of file LcsRowScanExecStream.cpp.

References CircularBuffer< T >::clear(), clumpDistance, clumpPos, clumpSize, currRidRun, deletedRidEos, filters, iFilterToInitialize, ConfluenceExecStream::inAccessors, LbmRidReader::init(), initializeSystemSampling(), inputRid, isFullScan, isSamplingRepeatable, MAXU, nextRid, nFilters, LcsRidRun::nRids, nRidsRead, numClumpsBuilt, LcsRowScanBaseExecStream::open(), producePending, readDeletedRid, repeatableSeed, CircularBufferIter< T >::reset(), ridReader, ridRunIter, LcsRowScanBaseExecStream::ridRuns, ridRunsBuilt, ridTupleData, SAMPLING_BERNOULLI, SAMPLING_SYSTEM, samplingMode, samplingRng, LcsRidRun::startRid, and tupleFound.

{
    LcsRowScanBaseExecStream::open(restart);
    producePending = false;
    tupleFound = false;
    nRidsRead = 0;
    ridRunsBuilt = false;
    currRidRun.startRid = LcsRid(MAXU);
    currRidRun.nRids = 0;
    ridRuns.clear();
    ridRunIter.reset();

    if (isFullScan) {
        inputRid = LcsRid(0);
        readDeletedRid = true;
        deletedRidEos = false;
    }
    nextRid = LcsRid(0);
    ridReader.init(inAccessors[0], ridTupleData);

    /*
     * Read from the 1st input, but only if we're not doing a restart.
     * Restarts can reuse the structures set up on the initial open
     * because the current assumption is that the residual filter
     * values don't change in between restarts.  If on restart, if a filter
     * wasn't completely initialized, then reinitialize it.
     */
    if (!restart) {
        iFilterToInitialize = 0;
    } else if (iFilterToInitialize < nFilters) {
        if (!filters[iFilterToInitialize]->filterDataInitialized) {
            filters[iFilterToInitialize]->filterData.clear();
        }
    }

    if (samplingMode == SAMPLING_BERNOULLI) {
        if (isSamplingRepeatable) {
            samplingRng->reseed(repeatableSeed);
        } else if (!restart) {
            samplingRng->reseed(static_cast<uint32_t>(time(0)));
        }
    } else if (samplingMode == SAMPLING_SYSTEM) {
        clumpSize = 0;
        clumpDistance = 0;
        clumpPos = 0;
        numClumpsBuilt = 0;

        initializeSystemSampling();
    }
}

ExecStreamResult LcsRowScanExecStream::execute

(

ExecStreamQuantum const &

quantum

)

[virtual]

Executes this stream.

Parameters:

quantum

governs the maximum amount of execution to perform

Returns:

code indicating reason execution ceased

Implements ExecStream.

Definition at line 397 of file LcsRowScanExecStream.cpp.

References CircularBufferIter< T >::done(), EXECRC_BUF_OVERFLOW, EXECRC_BUF_UNDERFLOW, EXECRC_EOS, EXECRC_QUANTUM_EXPIRED, EXECRC_YIELD, fillRidRunBuffer(), CircularBufferIter< T >::getCurrPos(), LcsClusterReader::getFetchRids(), initializeFiltersIfNeeded(), isFullScan, LCS_RID_COLUMN_ID, MAXU, LcsRowScanBaseExecStream::nClusters, nextRid, CircularBuffer< T >::nFreeSpace(), LcsRowScanBaseExecStream::nonClusterCols, nRidsRead, ExecStreamQuantum::nTuplesMax, opaqueToInt(), outputProj, outputTupleData, LcsRowScanBaseExecStream::pClusters, SingleOutputExecStream::pOutAccessor, producePending, TupleData::projectFrom(), LcsRowScanBaseExecStream::projMap, projOutputTupleData, LcsRowScanBaseExecStream::readColVals(), TupleDataWithBuffer::resetBuffer(), ridRunIter, LcsRowScanBaseExecStream::ridRuns, ridRunsBuilt, LcsRowScanBaseExecStream::syncColumns(), and tupleFound.

{
    if (!initializeFiltersIfNeeded()) {
        return EXECRC_BUF_UNDERFLOW;
    }

    for (uint i = 0; i < quantum.nTuplesMax; i++) {
        uint iClu;
        bool passedFilter;

        while (!producePending) {
            // No need to fill the rid run buffer each time through the loop
            if (!ridRunsBuilt && ridRuns.nFreeSpace() > 100) {
                ExecStreamResult rc = fillRidRunBuffer();
                if (rc != EXECRC_YIELD) {
                    return rc;
                }
            }

            // Determine the rid that needs to be fetched based on the
            // contents of the rid run buffer.
            LcsRid rid =
                LcsClusterReader::getFetchRids(ridRunIter, nextRid, true);
            if (rid == LcsRid(MAXU)) {
                assert(ridRunIter.done());
                pOutAccessor->markEOS();
                return EXECRC_EOS;
            }

            uint prevClusterEnd = 0;
            // reset datum pointers, in case previous tuple had nulls
            outputTupleData.resetBuffer();

            // Read the non-cluster columns first
            for (uint j = 0; j < nonClusterCols.size(); j++) {
                if (nonClusterCols[j] == LCS_RID_COLUMN_ID) {
                    memcpy(
                        const_cast<PBuffer>(outputTupleData[projMap[j]].pData),
                        (PBuffer) &rid, sizeof(LcsRid));
                    prevClusterEnd++;
                } else {
                    permAssert(false);
                }
            }

            // Then go through each cluster, forming rows and checking ranges
            for (iClu = 0, passedFilter = true; iClu <  nClusters; iClu++) {
                SharedLcsClusterReader &pScan = pClusters[iClu];

                // Resync the cluster reader to the current rid position
                pScan->catchUp(ridRunIter.getCurrPos(), nextRid);

                // if we have not read a batch yet or we've reached the
                // end of a batch, position to the rid we want to read

                if (!pScan->isPositioned() || rid >= pScan->getRangeEndRid()) {
                    bool rc = pScan->position(rid);

                    // rid not found, so just consume the rid and
                    // continue
                    if (rc == false)
                        break;

                    assert(rid >= pScan->getRangeStartRid()
                           && rid < pScan->getRangeEndRid());

                    // Tell all column scans that the batch has changed.
                    syncColumns(pScan);
                } else {
                    // Should not have moved into previous batch.
                    assert(rid > pScan->getRangeStartRid());

                    // move to correct position within scan; we know we
                    // will not fall off end of batch, so use non-checking
                    // function (for speed)
                    pScan->advanceWithinBatch(
                        opaqueToInt(rid - pScan->getCurrentRid()));
                }

                passedFilter =
                    readColVals(
                        pScan,
                        outputTupleData,
                        prevClusterEnd);
                if (!passedFilter) {
                    break;
                }
                prevClusterEnd += pScan->nColsToRead;
            }

            if (!passedFilter) {
                continue;
            }
            if (iClu == nClusters) {
                tupleFound = true;
            }
            producePending = true;
        }

        // produce tuple
        projOutputTupleData.projectFrom(outputTupleData, outputProj);
        if (tupleFound) {
            if (!pOutAccessor->produceTuple(projOutputTupleData)) {
                return EXECRC_BUF_OVERFLOW;
            }
        }
        producePending = false;

        if (isFullScan) {
            // if tuple not found, reached end of table
            if (!tupleFound) {
                pOutAccessor->markEOS();
                return EXECRC_EOS;
            }
        }

        tupleFound = false;
        nRidsRead++;
    }

    return EXECRC_QUANTUM_EXPIRED;
}

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

Reimplemented from LcsRowScanBaseExecStream.

Definition at line 249 of file LcsRowScanExecStream.cpp.

References LcsRowScanBaseExecStream::getResourceRequirements().

{
    LcsRowScanBaseExecStream::getResourceRequirements(minQuantity, optQuantity);
}

void LcsRowScanExecStream::closeImpl

(

)

[virtual]

Implements ClosableObject.

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

Reimplemented from LcsRowScanBaseExecStream.

Definition at line 672 of file LcsRowScanExecStream.cpp.

References LcsRowScanBaseExecStream::closeImpl(), filters, and nFilters.

{
    LcsRowScanBaseExecStream::closeImpl();

    for (uint i = 0; i < nFilters; i++) {
        filters[i]->filterData.clear();
    }
}

void LcsRowScanBaseExecStream::syncColumns

(

SharedLcsClusterReader &

pScan

)

[protected, inherited]

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(), execute(), LbmGeneratorExecStream::execute(), and LbmGeneratorExecStream::generateMultiKeyBitmaps().

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

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

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 LcsRowScanBaseExecStream::allSpecial, LcsRowScanBaseExecStream::attrAccessors, LcsRowScanBaseExecStream::projDescriptor, and LcsRowScanBaseExecStream::projMap.

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

{
    if (!allSpecial) {
        for (uint iCluCol = 0; iCluCol < pScan->nColsToRead; iCluCol++) {
            // Get value of each column and load it to the appropriate
            // tuple datum entry
            PBuffer curValue = pScan->clusterCols[iCluCol].getCurrentValue();
            uint idx = projMap[colStart + iCluCol];

            attrAccessors[idx].loadValue(tupleData[idx], curValue);
            if (pScan->clusterCols[iCluCol].getFilters().hasResidualFilters) {
                if (!pScan->clusterCols[iCluCol].applyFilters(
                    projDescriptor,
                    tupleData))
                {
                    return false;
                }
            }
        }
    }
    return true;
}

void LcsRowScanBaseExecStream::prepare

(

LcsRowScanBaseExecStreamParams const &

params

)

[virtual, inherited]

Definition at line 33 of file LcsRowScanBaseExecStream.cpp.

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

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

{
    ConfluenceExecStream::prepare(params);

    // Copy cluster definition parameters and setup btree readers for each
    // cluster.  Also, setup the full output tuple based on the ordered
    // list of cluster descriptors.

    nClusters = params.lcsClusterScanDefs.size();
    pClusters.reset(new SharedLcsClusterReader[nClusters]);

    uint clusterStart = 0;
    uint projCount = 0;
    TupleDescriptor allClusterTupleDesc;
    TupleProjection newProj, outputProj;

    buildOutputProj(outputProj, params);

    newProj.resize(outputProj.size());

    // if we're projecting non-cluster columns, keep track of them separately;
    TupleDescriptor outputTupleDesc = pOutAccessor->getTupleDesc();
    for (uint i = 0; i < outputProj.size(); i++) {
        if (outputProj[i] == LCS_RID_COLUMN_ID) {
            newProj[i] = projCount++;
            allClusterTupleDesc.push_back(outputTupleDesc[i]);
            nonClusterCols.push_back(params.outputProj[i]);
        }
    }

    allSpecial = (nonClusterCols.size() == newProj.size());

    for (uint i = 0; i < nClusters; i++) {
        SharedLcsClusterReader &pClu = pClusters[i];

        BTreeExecStreamParams const &bTreeParams = params.lcsClusterScanDefs[i];

        BTreeDescriptor treeDescriptor;
        treeDescriptor.segmentAccessor.pSegment = bTreeParams.pSegment;
        treeDescriptor.segmentAccessor.pCacheAccessor =
            bTreeParams.pCacheAccessor;
        treeDescriptor.tupleDescriptor = bTreeParams.tupleDesc;
        treeDescriptor.keyProjection = bTreeParams.keyProj;
        treeDescriptor.rootPageId = bTreeParams.rootPageId;
        treeDescriptor.segmentId = bTreeParams.segmentId;
        treeDescriptor.pageOwnerId = bTreeParams.pageOwnerId;

        pClu =
            SharedLcsClusterReader(
                new LcsClusterReader(treeDescriptor, &ridRuns));

        // setup the cluster and column readers to only read the columns
        // that are going to be projected
        uint clusterEnd = clusterStart +
            params.lcsClusterScanDefs[i].clusterTupleDesc.size() - 1;

        // create a vector of the columns that are projected from
        // this cluster and recompute the projection list
        // based on the individual cluster projections
        TupleProjection clusterProj;
        for (uint j = 0; j < newProj.size(); j++) {
            if (outputProj[j] >= clusterStart &&
                outputProj[j] <= clusterEnd)
            {
                clusterProj.push_back(outputProj[j] - clusterStart);
                newProj[j] = projCount++;
            }
        }
        clusterStart = clusterEnd + 1;

        // need to select at least one column from cluster, except in the
        // cases where we're only selecting special columns or when there
        // are filter columns; in the former case, we'll just arbitrarily
        // read the first column, but not actually project it
        if (allSpecial) {
           clusterProj.push_back(0);
        }
        pClu->initColumnReaders(
            params.lcsClusterScanDefs[i].clusterTupleDesc.size(),
            clusterProj);
        if (!allSpecial) {
            for (uint j = 0; j < pClu->nColsToRead; j++) {
                allClusterTupleDesc.push_back(
                    params.lcsClusterScanDefs[i].
                        clusterTupleDesc[clusterProj[j]]);
            }
        }
    }

    // setup projected tuple descriptor, by reshuffling allClusterTupleDesc
    // built above, into the correct projection order

    for (uint i = 0; i < newProj.size(); i++) {
        projDescriptor.push_back(allClusterTupleDesc[newProj[i]]);
    }

    // create a projection map to map cluster data read to the output
    // projection
    projMap.resize(newProj.size());
    for (uint i = 0; i < projMap.size(); i++) {
        for (uint j = 0; j < newProj.size(); j++) {
            if (newProj[j] == i) {
                projMap[i] = j;
            }
        }
    }
}

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

{
    SingleOutputExecStream::prepare(params);

    for (uint i = 0; i < inAccessors.size(); ++i) {
        assert(inAccessors[i]->getProvision() == getInputBufProvision());
    }
}

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

{
    ExecStream::prepare(params);
    assert(pOutAccessor);
    assert(pOutAccessor->getProvision() == getOutputBufProvision());
    if (pOutAccessor->getTupleDesc().empty()) {
        assert(!params.outputTupleDesc.empty());
        pOutAccessor->setTupleShape(
            params.outputTupleDesc,
            params.outputTupleFormat);
    }
}

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

{
    if (pGraph) {
        pDynamicParamManager = pGraph->getDynamicParamManager();
    }
    pQuotaAccessor = params.pCacheAccessor;
    pScratchQuotaAccessor = params.scratchAccessor.pCacheAccessor;
}

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.

Referenced by ExternalSortExecStreamImpl::getResourceRequirements(), LcsRowScanBaseExecStream::getResourceRequirements(), LcsClusterAppendExecStream::getResourceRequirements(), LbmUnionExecStream::getResourceRequirements(), LbmSplicerExecStream::getResourceRequirements(), LbmGeneratorExecStream::getResourceRequirements(), LbmChopperExecStream::getResourceRequirements(), LhxJoinExecStream::getResourceRequirements(), LhxAggExecStream::getResourceRequirements(), FtrsTableWriterExecStream::getResourceRequirements(), BTreeReadExecStream::getResourceRequirements(), BTreeInsertExecStream::getResourceRequirements(), FlatFileExecStreamImpl::getResourceRequirements(), SegBufferWriterExecStream::getResourceRequirements(), SegBufferReaderExecStream::getResourceRequirements(), SegBufferExecStream::getResourceRequirements(), ScratchBufferExecStream::getResourceRequirements(), and DoubleBufferExecStream::getResourceRequirements().

{
    getResourceRequirements(minQuantity, optQuantity);
    optType = EXEC_RESOURCE_ACCURATE;
}

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.

{
    inAccessors = inAccessorsInit;
}

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

{
    return BUFPROV_PRODUCER;
}

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

{
    assert(outAccessors.size() == 1);
    pOutAccessor = outAccessors[0];
}

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

{
    return BUFPROV_CONSUMER;
}

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.

{
    return true;
}

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

{
    assert(pGraph);
    return *pGraph;
}

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

{
    return id;
}

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

{
    resourceAllocation = quantity;
    if (pQuotaAccessor) {
        pQuotaAccessor->setMaxLockedPages(quantity.nCachePages);
    }
    if (pScratchQuotaAccessor) {
        pScratchQuotaAccessor->setMaxLockedPages(quantity.nCachePages);
    }
}

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.

{
    name = nameInit;
}

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

{
    return name;
}

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.

{
    return false;
}

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

{
    if (!pGraph) {
        return;
    }
    ExecStreamScheduler *pScheduler = pGraph->getScheduler();
    if (!pScheduler) {
        return;
    }
    pScheduler->checkAbort();
}

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.

{
    return BUFPROV_NONE;
}

bool ClosableObject::isClosed

(

)

const [inline, inherited]

Returns:

whether the object has been closed

Definition at line 58 of file ClosableObject.h.

    {
        return !needsClose;
    }

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

{
    if (!needsClose) {
        return;
    }
    needsClose = false;
    closeImpl();
}

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

{
    assert(!pTraceTarget.get());

    pTraceTarget = pTraceTargetInit;
    name = nameInit;
    if (isTracing()) {
        minimumLevel = pTraceTarget->getSourceTraceLevel(name);
    } else {
        minimumLevel = TRACE_OFF;
    }
}

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

{
    if (isTracing()) {
        getTraceTarget().notifyTrace(name,level,message);
    }
}

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

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

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

    {
        return level >= minimumLevel;
    }

TraceTarget& TraceSource::getTraceTarget

(

)

const [inline, inherited]

Returns:

the TraceTarget for this source

Definition at line 108 of file TraceSource.h.

Referenced by TraceSource::trace().

    {
        assert(isTracing());
        return *(pTraceTarget.get());
    }

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

    {
        return pTraceTarget;
    }

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

    {
        return name;
    }

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.

    {
        name = n;
    }

TraceLevel TraceSource::getMinimumTraceLevel

(

)

const [inline, inherited]

Definition at line 141 of file TraceSource.h.

    {
        return minimumLevel;
    }

void TraceSource::disableTracing

(

)

[inherited]

Definition at line 68 of file TraceSource.cpp.

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

Referenced by TestBase::afterTestCase().

{
    pTraceTarget.reset();
    minimumLevel = TRACE_OFF;
}

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

{
    pErrorTarget = pErrorTargetInit;
    name = nameInit;
}

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

{
    if (hasTarget()) {
        getErrorTarget().notifyError(
            name, level, message, address, capacity, index);
    }
}

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

{
    if (!hasTarget()) {
        return;
    }

    if (!pErrorBuf) {
        errorAccessor.compute(errorDesc);
        uint cbMax = errorAccessor.getMaxByteCount();
        pErrorBuf.reset(new FixedBuffer[cbMax]);
    }

    uint cbTuple = errorAccessor.getByteCount(errorTuple);
    errorAccessor.marshal(errorTuple, pErrorBuf.get());
    postError(level, message, pErrorBuf.get(), cbTuple, index);
}

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

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

ErrorTarget& ErrorSource::getErrorTarget

(

)

const [inline, inherited]

Returns:

the ErrorTarget for this source

Definition at line 120 of file ErrorSource.h.

Referenced by ErrorSource::postError().

    {
        assert(hasTarget());
        return *(pErrorTarget.get());
    }

SharedErrorTarget ErrorSource::getSharedErrorTarget

(

)

const [inline, inherited]

Returns:

the SharedErrorTarget for this source

Definition at line 129 of file ErrorSource.h.

    {
        return pErrorTarget;
    }

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.

    {
        return name;
    }

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.

    {
        name = n;
    }

void ErrorSource::disableTarget

(

)

[inherited]

Definition at line 84 of file ErrorSource.cpp.

References ErrorSource::pErrorTarget.

{
    pErrorTarget.reset();
}

---

Member Data Documentation

TupleDataWithBuffer LcsRowScanExecStream::outputTupleData [private]

Tuple data for all columns read from all clusters, including filter columns.

Definition at line 117 of file LcsRowScanExecStream.h.

Referenced by execute(), and prepare().

uint LcsRowScanExecStream::iFilterToInitialize [private]

This variable is used to control the initialization of residual filters.

It's 1 less than the index of the first filtering input to read. After open, it's initializaed to 0. On execute, the filtering inputs are read sequentially, while this variable is incremented, until an underflow or all filtering inputs have been read. On return due to an underflow, this variable allows reading to resume where it had left off.

Definition at line 129 of file LcsRowScanExecStream.h.

Referenced by initializeFiltersIfNeeded(), and open().

TupleData LcsRowScanExecStream::projOutputTupleData [private]

Definition at line 134 of file LcsRowScanExecStream.h.

Referenced by execute(), and prepare().

TupleProjection LcsRowScanExecStream::outputProj [private]

Definition at line 139 of file LcsRowScanExecStream.h.

Referenced by buildOutputProj(), execute(), and prepare().

TupleData LcsRowScanExecStream::ridTupleData [private]

Tuple data for input stream.

Definition at line 144 of file LcsRowScanExecStream.h.

Referenced by open(), and prepare().

LbmRidReader LcsRowScanExecStream::ridReader [private]

Rid reader.

Definition at line 149 of file LcsRowScanExecStream.h.

Referenced by fillRidRunBuffer(), and open().

RecordNum LcsRowScanExecStream::nRidsRead [private]

Number of rids read.

Definition at line 154 of file LcsRowScanExecStream.h.

Referenced by execute(), and open().

LcsRid LcsRowScanExecStream::inputRid [private]

Current rid read from the input stream.

Definition at line 159 of file LcsRowScanExecStream.h.

Referenced by fillRidRunBuffer(), and open().

LcsRid LcsRowScanExecStream::nextRid [private]

Next rid that needs to be fetched.

Definition at line 164 of file LcsRowScanExecStream.h.

Referenced by execute(), and open().

bool LcsRowScanExecStream::readDeletedRid [private]

True if need to read a new deleted rid from the input stream.

Definition at line 169 of file LcsRowScanExecStream.h.

Referenced by fillRidRunBuffer(), and open().

bool LcsRowScanExecStream::deletedRidEos [private]

True if reached EOS on deleted rid input stream.

Definition at line 174 of file LcsRowScanExecStream.h.

Referenced by fillRidRunBuffer(), and open().

LcsRid LcsRowScanExecStream::deletedRid [private]

Current deleted rid.

Definition at line 179 of file LcsRowScanExecStream.h.

Referenced by fillRidRunBuffer().

bool LcsRowScanExecStream::tupleFound [private]

true if tuple has been read and not yet produced

Definition at line 184 of file LcsRowScanExecStream.h.

Referenced by execute(), and open().

bool LcsRowScanExecStream::isFullScan [private]

true if executing full table scan

Definition at line 189 of file LcsRowScanExecStream.h.

Referenced by execute(), fillRidRunBuffer(), open(), and prepare().

bool LcsRowScanExecStream::hasExtraFilter [private]

true if there's extra range list filter(as the last input)

Definition at line 194 of file LcsRowScanExecStream.h.

Referenced by prepare().

bool LcsRowScanExecStream::producePending [private]

true if produceTuple pending

Definition at line 199 of file LcsRowScanExecStream.h.

Referenced by execute(), and open().

boost::scoped_array<LcsResidualColumnFilters *> LcsRowScanExecStream::filters [private]

The local filter data structure.

Note that these are aliasing pointers to facilitate filter data initialization and memory deallocation.

Definition at line 207 of file LcsRowScanExecStream.h.

Referenced by closeImpl(), initializeFiltersIfNeeded(), open(), and prepareResidualFilters().

int32_t LcsRowScanExecStream::nFilters [private]

The number of residual column filters configured.

Definition at line 212 of file LcsRowScanExecStream.h.

Referenced by closeImpl(), initializeFiltersIfNeeded(), open(), prepare(), and prepareResidualFilters().

TableSamplingMode LcsRowScanExecStream::samplingMode [private]

One of SAMPLING_OFF, SAMPLING_BERNOULLI or SAMPLING_SYSTEM.

Definition at line 217 of file LcsRowScanExecStream.h.

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

float LcsRowScanExecStream::samplingRate [private]

the sampling rate (0.0 to 1.0)

Definition at line 222 of file LcsRowScanExecStream.h.

Referenced by initializeSystemSampling(), and prepare().

bool LcsRowScanExecStream::isSamplingRepeatable [private]

true if the sample should be repeatable

Definition at line 227 of file LcsRowScanExecStream.h.

Referenced by open(), and prepare().

int32_t LcsRowScanExecStream::repeatableSeed [private]

seed for repeatable sampling

Definition at line 232 of file LcsRowScanExecStream.h.

Referenced by open(), and prepare().

int32_t LcsRowScanExecStream::samplingClumps [private]

number of clumps for system sampling

Definition at line 237 of file LcsRowScanExecStream.h.

Referenced by initializeSystemSampling(), and prepare().

uint64_t LcsRowScanExecStream::clumpSize [private]

size of each sampling clump

Definition at line 242 of file LcsRowScanExecStream.h.

Referenced by fillRidRunBuffer(), initializeSystemSampling(), and open().

uint64_t LcsRowScanExecStream::clumpDistance [private]

distance (in rows) between each clump

Definition at line 247 of file LcsRowScanExecStream.h.

Referenced by fillRidRunBuffer(), initializeSystemSampling(), and open().

uint64_t LcsRowScanExecStream::clumpPos [private]

position (0 to clumpSize) in current clump

Definition at line 252 of file LcsRowScanExecStream.h.

Referenced by fillRidRunBuffer(), initializeSystemSampling(), and open().

uint64_t LcsRowScanExecStream::clumpSkipPos [private]

position (clumpDistance to 0) in between clumps

Definition at line 257 of file LcsRowScanExecStream.h.

Referenced by fillRidRunBuffer(), and initializeSystemSampling().

uint LcsRowScanExecStream::numClumps [private]

The number of clumps that need to be built.

Definition at line 262 of file LcsRowScanExecStream.h.

Referenced by fillRidRunBuffer(), and initializeSystemSampling().

uint LcsRowScanExecStream::numClumpsBuilt [private]

Running counter of the number of clumps built.

Definition at line 267 of file LcsRowScanExecStream.h.

Referenced by fillRidRunBuffer(), and open().

boost::scoped_ptr<BernoulliRng> LcsRowScanExecStream::samplingRng [private]

RNG for Bernoulli sampling.

Definition at line 272 of file LcsRowScanExecStream.h.

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

int64_t LcsRowScanExecStream::rowCount [private]

Number of rows in the table.

Used only for sampling. In the case of Bernoulli sampling, includes count of deleted rows.

Definition at line 278 of file LcsRowScanExecStream.h.

Referenced by fillRidRunBuffer(), initializeSystemSampling(), and prepare().

bool LcsRowScanExecStream::ridRunsBuilt [private]

True if completed building rid runs.

Definition at line 283 of file LcsRowScanExecStream.h.

Referenced by execute(), fillRidRunBuffer(), and open().

LcsRidRun LcsRowScanExecStream::currRidRun [private]

Current rid run being constructed.

Definition at line 288 of file LcsRowScanExecStream.h.

Referenced by fillRidRunBuffer(), and open().

CircularBufferIter<LcsRidRun> LcsRowScanExecStream::ridRunIter [private]

Iterator over the circular buffer containing rid runs.

Definition at line 293 of file LcsRowScanExecStream.h.

Referenced by execute(), and open().

VectorOfUint LcsRowScanBaseExecStream::projMap [protected, inherited]

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 execute(), LbmGeneratorExecStream::generateSingletons(), LcsRowScanBaseExecStream::prepare(), prepareResidualFilters(), and LcsRowScanBaseExecStream::readColVals().

uint LcsRowScanBaseExecStream::nClusters [protected, inherited]

Number of clusters to be scanned.

Definition at line 82 of file LcsRowScanBaseExecStream.h.

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

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

Array containing cluster readers.

Definition at line 87 of file LcsRowScanBaseExecStream.h.

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

TupleDescriptor LcsRowScanBaseExecStream::projDescriptor [protected, inherited]

Tuple descriptor representing columns to be projected from scans.

Definition at line 92 of file LcsRowScanBaseExecStream.h.

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

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

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

Definition at line 97 of file LcsRowScanBaseExecStream.h.

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

bool LcsRowScanBaseExecStream::allSpecial [protected, inherited]

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 LcsRowScanBaseExecStream::prepare(), and LcsRowScanBaseExecStream::readColVals().

CircularBuffer<LcsRidRun> LcsRowScanBaseExecStream::ridRuns [protected, inherited]

Circular buffer of rid runs.

Definition at line 108 of file LcsRowScanBaseExecStream.h.

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

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

Accessors used for loading actual column values.

Definition at line 120 of file LcsRowScanBaseExecStream.h.

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

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

Definition at line 50 of file ConfluenceExecStream.h.

Referenced by NestedLoopJoinExecStream::checkNumInputs(), CartesianJoinExecStream::checkNumInputs(), LbmMinusExecStream::comparePrefixes(), LbmGeneratorExecStream::execute(), MergeExecStream::execute(), BarrierExecStream::execute(), LbmMinusExecStream::findMinInput(), initializeFiltersIfNeeded(), open(), LbmUnionExecStream::open(), LbmMinusExecStream::open(), LbmGeneratorExecStream::open(), LbmChopperExecStream::open(), LbmBitOpExecStream::open(), ConfluenceExecStream::open(), prepare(), LbmUnionExecStream::prepare(), LbmMinusExecStream::prepare(), LbmGeneratorExecStream::prepare(), LbmChopperExecStream::prepare(), LbmBitOpExecStream::prepare(), LhxJoinExecStream::prepare(), MergeExecStream::prepare(), CorrelationJoinExecStream::prepare(), ConfluenceExecStream::prepare(), CartesianJoinExecStream::prepare(), BarrierExecStream::prepare(), NestedLoopJoinExecStream::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.

Referenced by LcsClusterAppendExecStream::compress(), ExternalSortExecStreamImpl::execute(), execute(), LbmUnionExecStream::execute(), LbmNormalizerExecStream::execute(), LbmGeneratorExecStream::execute(), LbmChopperExecStream::execute(), LhxJoinExecStream::execute(), LhxAggExecStream::execute(), FtrsTableWriterExecStream::execute(), BTreeSortExecStream::execute(), BTreeSearchUniqueExecStream::execute(), BTreeScanExecStream::execute(), BTreePrefetchSearchExecStream::execute(), BTreeInsertExecStream::execute(), FlatFileExecStreamImpl::execute(), ValuesExecStream::execute(), UncollectExecStream::execute(), SortedAggExecStream::execute(), SegBufferReaderExecStream::execute(), ScratchBufferExecStream::execute(), ReshapeExecStream::execute(), MockResourceExecStream::execute(), MockProducerExecStream::execute(), MergeExecStream::execute(), DoubleBufferExecStream::execute(), CorrelationJoinExecStream::execute(), CopyExecStream::execute(), CollectExecStream::execute(), CartesianJoinExecStream::execute(), BernoulliSamplingExecStream::execute(), BarrierExecStream::execute(), CalcExecStream::execute(), LbmGeneratorExecStream::flushEntry(), MockProducerExecStream::getProducedRowCount(), ExternalSortExecStreamImpl::getResourceRequirements(), BTreeSearchExecStream::innerFetchLoop(), LbmUnionExecStream::open(), LbmChopperExecStream::open(), LbmBitOpExecStream::open(), SingleOutputExecStream::open(), SegBufferReaderExecStream::open(), SegBufferExecStream::open(), ScratchBufferExecStream::open(), DoubleBufferExecStream::open(), CollectExecStream::open(), SegBufferExecStream::openBufferForRead(), ConduitExecStream::precheckConduitBuffers(), prepare(), LcsRowScanBaseExecStream::prepare(), LcsClusterAppendExecStream::prepare(), LbmUnionExecStream::prepare(), LbmGeneratorExecStream::prepare(), LhxJoinExecStream::prepare(), LhxAggExecStream::prepare(), FtrsTableWriterExecStream::prepare(), FlatFileExecStreamImpl::prepare(), UncollectExecStream::prepare(), SortedAggExecStream::prepare(), SingleOutputExecStream::prepare(), ReshapeExecStream::prepare(), MockResourceExecStream::prepare(), MockProducerExecStream::prepare(), MergeExecStream::prepare(), CorrelationJoinExecStream::prepare(), ConduitExecStream::prepare(), CollectExecStream::prepare(), CartesianJoinExecStream::prepare(), BernoulliSamplingExecStream::prepare(), BarrierExecStream::prepare(), CalcExecStream::prepare(), SortedAggExecStream::produce(), LbmBitOpExecStream::producePendingOutput(), LbmUnionExecStream::produceTuple(), LbmNormalizerExecStream::produceTuple(), LbmMinusExecStream::produceTuple(), LbmChopperExecStream::produceTuple(), LbmBitOpExecStream::produceTuple(), LbmBitOpExecStream::readInput(), LbmMinusExecStream::readMinuendInput(), and SingleOutputExecStream::setOutputBufAccessors().

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.

Referenced by CorrelationJoinExecStream::close(), SegBufferReaderExecStream::closeImpl(), LbmUnionExecStream::execute(), LbmSplicerExecStream::execute(), LbmGeneratorExecStream::execute(), LbmChopperExecStream::execute(), CorrelationJoinExecStream::execute(), BarrierExecStream::execute(), LcsClusterReplaceExecStream::open(), LbmUnionExecStream::open(), LbmSplicerExecStream::open(), LbmGeneratorExecStream::open(), LbmBitOpExecStream::open(), BTreeSearchExecStream::open(), BTreeInsertExecStream::open(), SegBufferWriterExecStream::open(), SegBufferReaderExecStream::open(), NestedLoopJoinExecStream::open(), CorrelationJoinExecStream::open(), ExecStream::prepare(), CalcExecStream::prepare(), NestedLoopJoinExecStream::processLeftInput(), LbmSearchExecStream::reachedTupleLimit(), ReshapeExecStream::readDynamicParams(), SegBufferWriterExecStream::readReaderRefCount(), BTreeSearchExecStream::readSearchKey(), BTreeSearchExecStream::readUpperBoundKey(), LbmSearchExecStream::setAdditionalKeys(), and LbmBitOpExecStream::writeStartRidParamValue().

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

ExecStreamResourceQuantity ExecStream::resourceAllocation [protected, inherited]

Resource quantities currently allocated to this stream.

Definition at line 100 of file ExecStream.h.

Referenced by ExecStream::setResourceAllocation().

SharedCacheAccessor ExecStream::pQuotaAccessor [protected, inherited]

CacheAccessor used for quota tracking.

Definition at line 105 of file ExecStream.h.

Referenced by ExecStream::open(), ExecStream::prepare(), and ExecStream::setResourceAllocation().

SharedCacheAccessor ExecStream::pScratchQuotaAccessor [protected, inherited]

CacheAccessor used for scratch page quota tracking.

Definition at line 110 of file ExecStream.h.

Referenced by ExecStream::open(), ExecStream::prepare(), and ExecStream::setResourceAllocation().

bool ClosableObject::needsClose [protected, inherited]

Definition at line 44 of file ClosableObject.h.

Referenced by SegStreamAllocation::beginWrite(), ExecStreamGraphImpl::clear(), ClosableObject::ClosableObject(), ClosableObject::close(), FlatFileBuffer::open(), ExecStreamGraphImpl::open(), ExecStream::open(), and ClosableObject::~ClosableObject().

---

The documentation for this class was generated from the following files:

• /home/pub/open/dev/fennel/lucidera/colstore/LcsRowScanExecStream.h
• /home/pub/open/dev/fennel/lucidera/colstore/LcsRowScanExecStream.cpp

---