Fennel: SortedAggExecStream Class Reference

enum SortedAggExecStream::State [private]

Enumerator:

STATE_ACCUMULATING
STATE_PRODUCING
STATE_DONE

Definition at line 58 of file SortedAggExecStream.h.

00058                {
00059         STATE_ACCUMULATING,
00060         STATE_PRODUCING,
00061         STATE_DONE
00062     };

void SortedAggExecStream::clearAccumulator ( ) [inline, private]

Definition at line 106 of file SortedAggExecStream.cpp.

References aggComputers, groupByKeyCount, and prevTuple.

Referenced by open(), and produce().

00107 {
00108     for (int i = 0; i < aggComputers.size(); ++i) {
00109         aggComputers[i].clearAccumulator(prevTuple[i + groupByKeyCount]);
00110     }
00111 }

void SortedAggExecStream::updateAccumulator ( ) [inline, private]

Definition at line 113 of file SortedAggExecStream.cpp.

References aggComputers, groupByKeyCount, inputTuple, and prevTuple.

Referenced by execute(), and produce().

00114 {
00115     for (int i = 0; i < aggComputers.size(); ++i) {
00116         aggComputers[i].updateAccumulator(
00117             prevTuple[i + groupByKeyCount],
00118             inputTuple);
00119     }
00120 }

void SortedAggExecStream::computeOutput ( ) [inline, private]

Definition at line 150 of file SortedAggExecStream.cpp.

References aggComputers, groupByKeyCount, outputTuple, and prevTuple.

Referenced by execute().

00151 {
00152     int i;
00153 
00154     for (i = 0; i < groupByKeyCount; i ++) {
00155         outputTuple[i] = prevTuple[i];
00156     }
00157 
00158     for (i = 0; i < aggComputers.size(); i ++) {
00159         aggComputers[i].computeOutput(
00160             outputTuple[i + groupByKeyCount],
00161             prevTuple[i + groupByKeyCount]);
00162     }
00163 }

void SortedAggExecStream::copyPrevGroupByKey ( ) [inline, private]

Definition at line 122 of file SortedAggExecStream.cpp.

References groupByKeyCount, inputTuple, prevTuple, and TupleDataWithBuffer::resetBuffer().

Referenced by execute(), and produce().

00123 {
00124     /*
00125       Need to make sure pointers are allocated before memcpy.
00126       resetBuffer restores the pointers to the associated buffer.
00127     */
00128     prevTuple.resetBuffer();
00129 
00130     for (int i = 0; i < groupByKeyCount; i ++) {
00131         prevTuple[i].memCopyFrom(inputTuple[i]);
00132     }
00133 }

void SortedAggExecStream::setCurGroupByKey ( ) [inline, private]

int SortedAggExecStream::compareGroupByKeys ( ) [inline, private]

Definition at line 135 of file SortedAggExecStream.cpp.

References groupByKeyCount, inputTuple, SingleInputExecStream::pInAccessor, and prevTuple.

Referenced by execute().

00136 {
00137     /*
00138       prevTuple does not actually have the same Tuple layout
00139       as inputTuple; however, the prefixes(of length groupByKeyCount)
00140       refer to the same fields. Compare only the prefixes.
00141     */
00142     int ret =
00143         (pInAccessor->getTupleDesc()).compareTuplesKey(
00144             prevTuple,
00145             inputTuple,
00146             groupByKeyCount);
00147     return ret;
00148 }

ExecStreamResult SortedAggExecStream::produce ( ) [inline, private]

Definition at line 185 of file SortedAggExecStream.cpp.

References clearAccumulator(), copyPrevGroupByKey(), EXECRC_BUF_OVERFLOW, EXECRC_YIELD, outputTuple, SingleInputExecStream::pInAccessor, SingleOutputExecStream::pOutAccessor, state, STATE_ACCUMULATING, STATE_PRODUCING, and updateAccumulator().

Referenced by execute().

00186 {
00187     assert (state == STATE_PRODUCING);
00188 
00189     // attempt to write output
00190     bool success = pOutAccessor->produceTuple(outputTuple);
00191     if (success) {
00192         clearAccumulator();
00193         state = STATE_ACCUMULATING;
00194         /*
00195           Succeeded in outputting result for the previous group.
00196           Record new group by key and update accumulator result fields.
00197         */
00198         copyPrevGroupByKey();
00199         updateAccumulator();
00200         pInAccessor->consumeTuple();
00201         return EXECRC_YIELD;
00202     } else {
00203         return EXECRC_BUF_OVERFLOW;
00204     }
00205 }

AggComputer * SortedAggExecStream::newAggComputer	(	AggFunction	aggFunction,
		TupleAttributeDescriptor const *	pAttrDesc
	)			`[protected, virtual]`

Reimplemented in LbmSortedAggExecStream.

Definition at line 99 of file SortedAggExecStream.cpp.

References AggComputer::newAggComputer().

Referenced by LbmSortedAggExecStream::newAggComputer(), and prepare().

00102 {
00103     return AggComputer::newAggComputer(aggFunction, pAttrDesc);
00104 }

void SortedAggExecStream::prepare ( SortedAggExecStreamParams const & params ) [virtual]

Definition at line 30 of file SortedAggExecStream.cpp.

References AGG_FUNC_COUNT, AGG_FUNC_MAX, AGG_FUNC_MIN, AGG_FUNC_SINGLE_VALUE, AGG_FUNC_SUM, aggComputers, SortedAggExecStreamParams::aggInvocations, TupleData::compute(), TupleDataWithBuffer::computeAndAllocate(), SortedAggExecStreamParams::groupByKeyCount, groupByKeyCount, inputTuple, newAggComputer(), StandardTypeDescriptorFactory::newDataType(), outputTuple, SingleInputExecStream::pInAccessor, SingleOutputExecStream::pOutAccessor, ConduitExecStream::prepare(), prevTuple, and STANDARD_TYPE_INT_64.

Referenced by LbmSortedAggExecStream::prepare().

00031 {
00032     ConduitExecStream::prepare(params);
00033 
00034     inputTuple.compute(pInAccessor->getTupleDesc());
00035 
00036     /*
00037       prevTuple contains the groupByKey fields as well as the accumulator
00038       result fields. outputTuple has the same format as prevTuple.
00039       The difference is that prevTuple has buffer associated with it while
00040       outputTuple has pointers pointing to the result location.
00041     */
00042     TupleDescriptor prevTupleDesc;
00043     TupleDescriptor const &inputDesc = pInAccessor->getTupleDesc();
00044 
00045     // Attribute descriptor for COUNT output
00046     StandardTypeDescriptorFactory stdTypeFactory;
00047     TupleAttributeDescriptor countDesc(
00048         stdTypeFactory.newDataType(STANDARD_TYPE_INT_64));
00049 
00050     groupByKeyCount = params.groupByKeyCount;
00051 
00052     for (int i = 0; i < groupByKeyCount; i ++) {
00053         prevTupleDesc.push_back(inputDesc[i]);
00054     }
00055 
00056     /*
00057       Compute the accumulator result portion of prevTupleDesc based on
00058       requested aggregate function invocations, and instantiate polymorphic
00059       AggComputers bound to correct inputs.
00060     */
00061     for (AggInvocationConstIter pInvocation(params.aggInvocations.begin());
00062          pInvocation != params.aggInvocations.end();
00063          ++pInvocation)
00064     {
00065         switch (pInvocation->aggFunction) {
00066         case AGG_FUNC_COUNT:
00067             prevTupleDesc.push_back(countDesc);
00068             break;
00069         case AGG_FUNC_SUM:
00070         case AGG_FUNC_MIN:
00071         case AGG_FUNC_MAX:
00072         case AGG_FUNC_SINGLE_VALUE:
00073             // Output type is same as input type, but nullable
00074             prevTupleDesc.push_back(inputDesc[pInvocation->iInputAttr]);
00075             prevTupleDesc.back().isNullable = true;
00076             break;
00077         }
00078         TupleAttributeDescriptor const *pInputAttr = NULL;
00079         if (pInvocation->iInputAttr != -1) {
00080             pInputAttr = &(inputDesc[pInvocation->iInputAttr]);
00081         }
00082         aggComputers.push_back(
00083             newAggComputer(
00084                 pInvocation->aggFunction,
00085                 pInputAttr));
00086         aggComputers.back().setInputAttrIndex(pInvocation->iInputAttr);
00087     }
00088 
00089     // Sanity check:  the output shape we computed should agree with
00090     // the descriptor (if any) in the supplied plan.
00091     if (!params.outputTupleDesc.empty()) {
00092         assert(prevTupleDesc == params.outputTupleDesc);
00093     }
00094     prevTuple.computeAndAllocate(prevTupleDesc);
00095     outputTuple.compute(prevTupleDesc);
00096     pOutAccessor->setTupleShape(prevTupleDesc);
00097 }

void SortedAggExecStream::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 ConduitExecStream.

Definition at line 165 of file SortedAggExecStream.cpp.

References clearAccumulator(), groupByKeyCount, ConduitExecStream::open(), prevTupleValid, state, and STATE_ACCUMULATING.

00166 {
00167     ConduitExecStream::open(restart);
00168     clearAccumulator();
00169 
00170     /*
00171       When accumulating, the first tuple in a group always updates the
00172       accumulator result field. Compare the group by key fields only for
00173       subsequent tuples. We use prevTupleValid to mark if the first tuple
00174       was seen.
00175       Need to set this in open() so that when the same stream is re-executed,
00176       e.g. when two identical group by statements are issued, the state when
00177       receiving the first input tuple of the first group is correct.
00178       Ignore prevTupleValid field when not doing groupby's.
00179     */
00180     prevTupleValid = (groupByKeyCount > 0) ? false : true;
00181 
00182     state = STATE_ACCUMULATING;
00183 }

ExecStreamResult SortedAggExecStream::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 207 of file SortedAggExecStream.cpp.

References compareGroupByKeys(), computeOutput(), copyPrevGroupByKey(), EXECBUF_EOS, EXECRC_BUF_OVERFLOW, EXECRC_BUF_UNDERFLOW, EXECRC_EOS, EXECRC_YIELD, inputTuple, ExecStreamQuantum::nTuplesMax, outputTuple, SingleInputExecStream::pInAccessor, SingleOutputExecStream::pOutAccessor, ConduitExecStream::precheckConduitBuffers(), prevTupleValid, produce(), state, STATE_ACCUMULATING, STATE_DONE, STATE_PRODUCING, and updateAccumulator().

00208 {
00209     int keyComp;
00210     ExecStreamResult rc;
00211 
00212     /*
00213       Perform EOS processing first, since there can be a result tuple which is
00214       not produced yet.
00215     */
00216     if (pInAccessor->getState() == EXECBUF_EOS) {
00217         if (!prevTupleValid) {
00218             state = STATE_DONE;
00219         }
00220 
00221         // no more input is coming
00222         if (state == STATE_DONE) {
00223             // already produced output
00224             pOutAccessor->markEOS();
00225             return EXECRC_EOS;
00226         }
00227 
00228         if (state == STATE_ACCUMULATING) {
00229             // compute final output and get ready to write it
00230             computeOutput();
00231             state = STATE_PRODUCING;
00232         }
00233 
00234         // attempt to write output
00235         bool success = pOutAccessor->produceTuple(outputTuple);
00236         if (success) {
00237             state = STATE_DONE;
00238             // let precheckConduitBuffers below return EOS for us
00239         } else {
00240             return EXECRC_BUF_OVERFLOW;
00241         }
00242     } else if (state == STATE_PRODUCING) {
00243         rc = produce();
00244         if (rc != EXECRC_YIELD) {
00245             return rc;
00246         }
00247     }
00248 
00249     /*
00250       Check buffer state. If it is in a good state(EXECRC_YIELD, i.e. not in
00251       any abnormal state and is not empty),  process the tuples from the buffer.
00252     */
00253     rc = precheckConduitBuffers();
00254     if (rc != EXECRC_YIELD) {
00255         return rc;
00256     }
00257 
00258     /*
00259       Iterate through all the INPUT tuples. In this method, quantum represents
00260       unit of input data.
00261     */
00262     for (uint nTuples = 0; nTuples < quantum.nTuplesMax; ++nTuples) {
00263         if (!pInAccessor->demandData()) {
00264             return EXECRC_BUF_UNDERFLOW;
00265         }
00266 
00267         assert (state == STATE_ACCUMULATING);
00268 
00269         pInAccessor->unmarshalTuple(inputTuple);
00270 
00271         if (prevTupleValid) {
00272             keyComp = compareGroupByKeys();
00273             assert(keyComp <= 0);
00274             if (keyComp == 0) {
00275                 // continue reading rows and computing aggregates
00276                 // for this group
00277                 updateAccumulator();
00278                 pInAccessor->consumeTuple();
00279             } else {
00280                 // ready to produce an output row below
00281                 computeOutput();
00282                 state = STATE_PRODUCING;
00283             }
00284         } else {
00285             /*
00286               first tuple read so nothing to compare it to yet, but still need
00287               to record group by key and compute aggregates for that first row.
00288             */
00289             prevTupleValid = true;
00290             copyPrevGroupByKey();
00291             updateAccumulator();
00292             pInAccessor->consumeTuple();
00293         }
00294 
00295         if (state == STATE_PRODUCING) {
00296             rc = produce();
00297             if (rc != EXECRC_YIELD) {
00298                 return rc;
00299             }
00300         }
00301     }
00302 
00303     return EXECRC_YIELD;
00304 }

ExecStreamResult ConduitExecStream::precheckConduitBuffers ( ) [protected, inherited]

Checks the state of the input and output buffers.

If input empty, requests production. If input EOS, propagates that to output buffer. If output full, returns EXECRC_OVERFLOW.

Returns:: result of precheck; anything but EXECRC_YIELD indicates that execution should terminate immediately with returned code

Definition at line 61 of file ConduitExecStream.cpp.

References EXECBUF_EMPTY, EXECBUF_EOS, EXECBUF_NONEMPTY, EXECBUF_OVERFLOW, EXECBUF_UNDERFLOW, EXECRC_BUF_OVERFLOW, EXECRC_BUF_UNDERFLOW, EXECRC_EOS, EXECRC_YIELD, SingleInputExecStream::pInAccessor, and SingleOutputExecStream::pOutAccessor.

Referenced by ExternalSortExecStreamImpl::execute(), FtrsTableWriterExecStream::execute(), BTreeSearchUniqueExecStream::execute(), BTreeSearchExecStream::execute(), BTreeInsertExecStream::execute(), UncollectExecStream::execute(), execute(), ReshapeExecStream::execute(), CopyExecStream::execute(), CollectExecStream::execute(), BernoulliSamplingExecStream::execute(), and CalcExecStream::execute().

00062 {
00063     switch (pInAccessor->getState()) {
00064     case EXECBUF_EMPTY:
00065         pInAccessor->requestProduction();
00066         return EXECRC_BUF_UNDERFLOW;
00067     case EXECBUF_UNDERFLOW:
00068         return EXECRC_BUF_UNDERFLOW;
00069     case EXECBUF_EOS:
00070         pOutAccessor->markEOS();
00071         return EXECRC_EOS;
00072     case EXECBUF_NONEMPTY:
00073     case EXECBUF_OVERFLOW:
00074         break;
00075     default:
00076         permAssert(false);
00077     }
00078     if (pOutAccessor->getState() == EXECBUF_OVERFLOW) {
00079         return EXECRC_BUF_OVERFLOW;
00080     }
00081     return EXECRC_YIELD;
00082 }

void ConduitExecStream::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

Reimplemented from SingleInputExecStream.

Definition at line 36 of file ConduitExecStream.cpp.

References SingleOutputExecStream::setOutputBufAccessors().

00038 {
00039     SingleOutputExecStream::setOutputBufAccessors(outAccessors);
00040 }

void ConduitExecStream::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 SingleInputExecStream.

Definition at line 30 of file ConduitExecStream.cpp.

References SingleInputExecStream::setInputBufAccessors().

00032 {
00033     SingleInputExecStream::setInputBufAccessors(inAccessors);
00034 }

void ConduitExecStream::prepare ( ConduitExecStreamParams const & params ) [virtual, inherited]

Definition at line 42 of file ConduitExecStream.cpp.

References SingleInputExecStream::pInAccessor, SingleOutputExecStream::pOutAccessor, SingleOutputExecStream::prepare(), and SingleInputExecStream::prepare().

00043 {
00044     SingleInputExecStream::prepare(params);
00045 
00046     if (params.outputTupleDesc.empty()) {
00047         pOutAccessor->setTupleShape(
00048             pInAccessor->getTupleDesc(),
00049             pInAccessor->getTupleFormat());
00050     }
00051 
00052     SingleOutputExecStream::prepare(params);
00053 }

void SingleInputExecStream::prepare ( SingleInputExecStreamParams const & params ) [virtual, inherited]

Definition at line 44 of file SingleInputExecStream.cpp.

References SingleInputExecStream::getInputBufProvision(), SingleInputExecStream::pInAccessor, and ExecStream::prepare().

Referenced by JavaSinkExecStream::prepare(), MockConsumerExecStream::prepare(), DiffluenceExecStream::prepare(), and ConduitExecStream::prepare().

00045 {
00046     ExecStream::prepare(params);
00047 
00048     assert(pInAccessor);
00049     assert(pInAccessor->getProvision() == getInputBufProvision());
00050 }

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

ExecStreamBufProvision SingleInputExecStream::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.

Reimplemented in DoubleBufferExecStream, ScratchBufferExecStream, SegBufferExecStream, and SegBufferWriterExecStream.

Definition at line 62 of file SingleInputExecStream.cpp.

References BUFPROV_PRODUCER.

Referenced by SingleInputExecStream::prepare().

00063 {
00064     return BUFPROV_PRODUCER;
00065 }

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

Implements ClosableObject.

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

Implements ClosableObject.

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

Definition at line 54 of file ExecStream.cpp.

References ExecStream::isOpen.

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

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

Returns:: true if the stream can be closed early

Reimplemented in SegBufferWriterExecStream.

Definition at line 49 of file ExecStream.cpp.

00050 {
00051     return true;
00052 }

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

Returns:: reference to containing graph

Definition at line 293 of file ExecStream.h.

References ExecStream::pGraph.

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

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

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

Returns:: the identifier for this stream within containing graph

Definition at line 288 of file ExecStream.h.

References ExecStream::id.

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

00289 {
00290     return id;
00291 }

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

Determines resource requirements for this stream.

Default implementation declares zero resource requirements.

Parameters:

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

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

Definition at line 93 of file ExecStream.cpp.

References EXEC_RESOURCE_ACCURATE.

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

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

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

Definition at line 102 of file ExecStream.cpp.

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

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

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

Sets current resource allocation for this stream.

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

Parameters:

quantity

allocated resource quantity

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

Definition at line 111 of file ExecStream.cpp.

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

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

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

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

Sets unique name of this stream.

Definition at line 157 of file ExecStream.cpp.

References ExecStream::name.

00158 {
00159     name = nameInit;
00160 }

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

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

Definition at line 162 of file ExecStream.cpp.

References ExecStream::name.

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

00163 {
00164     return name;
00165 }

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

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

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

Returns:: whether stream may block; default is false

Definition at line 167 of file ExecStream.cpp.

00168 {
00169     return false;
00170 }

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

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

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

Definition at line 72 of file ExecStream.cpp.

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

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

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

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

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

Returns:: supported model; default is BUFPROV_NONE

Reimplemented in BarrierExecStream, DiffluenceExecStream, DoubleBufferExecStream, MergeExecStream, MockResourceExecStream, ScratchBufferExecStream, SegBufferExecStream, SegBufferReaderExecStream, SegBufferWriterExecStream, SingleOutputExecStream, SplitterExecStream, ValuesExecStream, JavaTransformExecStream, FtrsTableWriterExecStream, and LcsClusterAppendExecStream.

Definition at line 172 of file ExecStream.cpp.

References BUFPROV_NONE.

00173 {
00174     return BUFPROV_NONE;
00175 }

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 }

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

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

Returns:: required conversion; default is BUFPROV_NONE

Reimplemented in JavaTransformExecStream.

Definition at line 177 of file ExecStream.cpp.

References BUFPROV_NONE.

00178 {
00179     return BUFPROV_NONE;
00180 }

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

Returns:: whether the object has been closed

Definition at line 58 of file ClosableObject.h.

00059     {
00060         return !needsClose;
00061     }

void ClosableObject::close ( ) [inherited]

Closes this object, releasing any unallocated resources.

Reimplemented in CollectExecStream, CorrelationJoinExecStream, LcsClusterAppendExecStream, and LcsClusterReplaceExecStream.

Definition at line 39 of file ClosableObject.cpp.

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

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

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

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

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

Parameters:

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

Definition at line 46 of file TraceSource.cpp.

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

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

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

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

Records a trace message.

Normally only called via FENNEL_TRACE.

Parameters:

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

Definition at line 61 of file TraceSource.cpp.

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

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

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

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

Returns:: true iff tracing is enabled for this source

Definition at line 88 of file TraceSource.h.

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

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

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

Determines whether a particular level is being traced.

Parameters:

level

trace level to test

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

Definition at line 100 of file TraceSource.h.

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

00101     {
00102         return level >= minimumLevel;
00103     }

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

Returns:: the TraceTarget for this source

Definition at line 108 of file TraceSource.h.

Referenced by TraceSource::trace().

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

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

Returns:: the SharedTraceTarget for this source

Definition at line 117 of file TraceSource.h.

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

00118     {
00119         return pTraceTarget;
00120     }

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

Gets the name of this source.

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

Returns:: the name

Definition at line 127 of file TraceSource.h.

Referenced by LcsClusterAppendExecStream::initLoad().

00128     {
00129         return name;
00130     }

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

Sets the name of this source.

Useful to construct dynamic names for fine-grained filtering.

Definition at line 136 of file TraceSource.h.

00137     {
00138         name = n;
00139     }

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

Definition at line 141 of file TraceSource.h.

00142     {
00143         return minimumLevel;
00144     }

void TraceSource::disableTracing ( ) [inherited]

Definition at line 68 of file TraceSource.cpp.

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

Referenced by TestBase::afterTestCase().

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

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

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

Parameters:

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

Definition at line 47 of file ErrorSource.cpp.

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

Referenced by ErrorSource::ErrorSource().

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

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

Posts an exception, such as a row exception.

See also:: ErrorTarget for a description of the parameters

Definition at line 55 of file ErrorSource.cpp.

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

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

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

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

Posts an exception, such as a row exception.

See also:: ErrorTarget for a description of the parameters

Definition at line 65 of file ErrorSource.cpp.

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

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

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

Returns:: true iff an error target has been set

Definition at line 112 of file ErrorSource.h.

Referenced by ErrorSource::postError().

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

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

Returns:: the ErrorTarget for this source

Definition at line 120 of file ErrorSource.h.

Referenced by ErrorSource::postError().

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

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

Returns:: the SharedErrorTarget for this source

Definition at line 129 of file ErrorSource.h.

00130     {
00131         return pErrorTarget;
00132     }

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

Gets the name of this source.

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

Returns:: the name

Definition at line 139 of file ErrorSource.h.

00140     {
00141         return name;
00142     }

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

Sets the name of this source.

Useful to construct dynamic names for fine-grained filtering.

Definition at line 148 of file ErrorSource.h.

00149     {
00150         name = n;
00151     }

void ErrorSource::disableTarget ( ) [inherited]

Definition at line 84 of file ErrorSource.cpp.

References ErrorSource::pErrorTarget.

00085 {
00086     pErrorTarget.reset();
00087 }

State SortedAggExecStream::state [private]

Definition at line 64 of file SortedAggExecStream.h.

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

AggComputerList SortedAggExecStream::aggComputers [private]

Definition at line 66 of file SortedAggExecStream.h.

Referenced by clearAccumulator(), computeOutput(), prepare(), and updateAccumulator().

int SortedAggExecStream::groupByKeyCount [private]

Definition at line 67 of file SortedAggExecStream.h.

Referenced by clearAccumulator(), compareGroupByKeys(), computeOutput(), copyPrevGroupByKey(), open(), prepare(), and updateAccumulator().

TupleData SortedAggExecStream::inputTuple [private]

Definition at line 69 of file SortedAggExecStream.h.

Referenced by compareGroupByKeys(), copyPrevGroupByKey(), execute(), prepare(), and updateAccumulator().

TupleDataWithBuffer SortedAggExecStream::prevTuple [private]

Definition at line 70 of file SortedAggExecStream.h.

Referenced by clearAccumulator(), compareGroupByKeys(), computeOutput(), copyPrevGroupByKey(), prepare(), and updateAccumulator().

TupleData SortedAggExecStream::outputTuple [private]

Definition at line 71 of file SortedAggExecStream.h.

Referenced by computeOutput(), execute(), prepare(), and produce().

bool SortedAggExecStream::prevTupleValid [private]

Definition at line 72 of file SortedAggExecStream.h.

Referenced by execute(), and open().

SharedExecStreamBufAccessor SingleInputExecStream::pInAccessor [protected, inherited]

Definition at line 51 of file SingleInputExecStream.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().

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

SharedExecStreamBufAccessor SingleOutputExecStream::pOutAccessor [protected, inherited]

Definition at line 56 of file SingleOutputExecStream.h.

SortedAggExecStream Class Reference

Public Member Functions

Protected Member Functions

Protected Attributes

Private Types

Private Member Functions

Private Attributes

Detailed Description

Member Enumeration Documentation

Member Function Documentation

Member Data Documentation


Public Member Functions
virtual void	prepare (SortedAggExecStreamParams 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	setOutputBufAccessors (std::vector< SharedExecStreamBufAccessor > const &outAccessors)
	Initializes the buffer accessors for outputs from this stream.
virtual void	setInputBufAccessors (std::vector< SharedExecStreamBufAccessor > const &inAccessors)
	Initializes the buffer accessors for inputs to this stream.
virtual void	prepare (ConduitExecStreamParams const &params)
virtual void	prepare (SingleInputExecStreamParams const &params)
virtual void	prepare (ExecStreamParams const &params)
	Prepares this stream for execution.
virtual void	prepare (SingleOutputExecStreamParams const &params)
virtual ExecStreamBufProvision	getInputBufProvision () const
	Queries the BufferProvision which this stream requires of its inputs when consuming their tuples.
virtual bool	canEarlyClose ()
	Returns: true if the stream can be closed early
ExecStreamGraph &	getGraph () const
	Returns: reference to containing graph
ExecStreamId	getStreamId () const
	Returns: the identifier for this stream within containing graph
virtual void	getResourceRequirements (ExecStreamResourceQuantity &minQuantity, ExecStreamResourceQuantity &optQuantity, ExecStreamResourceSettingType &optType)
	Determines resource requirements for this stream.
virtual void	getResourceRequirements (ExecStreamResourceQuantity &minQuantity, ExecStreamResourceQuantity &optQuantity)
virtual void	setResourceAllocation (ExecStreamResourceQuantity &quantity)
	Sets current resource allocation for this stream.
virtual void	setName (std::string const &)
	Sets unique name of this stream.
virtual std::string const &	getName () const
	Returns: the name of this stream, as known by the optimizer
virtual bool	mayBlock () const
	Queries whether this stream's implementation may block when execute() is called.
virtual void	checkAbort () const
	Checks whether there is an abort request for this stream's scheduler.
virtual ExecStreamBufProvision	getOutputBufProvision () const
	Queries the BufferProvision which this stream is capable of when producing tuples.
virtual ExecStreamBufProvision	getOutputBufProvision () const
	Queries the BufferProvision which this stream is capable of when producing tuples.
virtual ExecStreamBufProvision	getOutputBufConversion () const
	Queries the BufferProvision to which this stream needs its output to be converted, if any.
bool	isClosed () const
	Returns: whether the object has been closed
void	close ()
	Closes this object, releasing any unallocated resources.
virtual void	initTraceSource (SharedTraceTarget pTraceTarget, std::string name)
	For use when initialization has to be deferred until after construction.
void	trace (TraceLevel level, std::string message) const
	Records a trace message.
bool	isTracing () const
	Returns: true iff tracing is enabled for this source
bool	isTracingLevel (TraceLevel level) const
	Determines whether a particular level is being traced.
TraceTarget &	getTraceTarget () const
	Returns: the TraceTarget for this source
SharedTraceTarget	getSharedTraceTarget () const
	Returns: the SharedTraceTarget for this source
std::string	getTraceSourceName () const
	Gets the name of this source.
void	setTraceSourceName (std::string const &n)
	Sets the name of this source.
TraceLevel	getMinimumTraceLevel () const
void	disableTracing ()
virtual void	initErrorSource (SharedErrorTarget pErrorTarget, const std::string &name)
	For use when initialization has to be deferred until after construction.
void	postError (ErrorLevel level, const std::string &message, void *address, long capacity, int index)
	Posts an exception, such as a row exception.
void	postError (ErrorLevel level, const std::string &message, const TupleDescriptor &errorDesc, const TupleData &errorTuple, int index)
	Posts an exception, such as a row exception.
bool	hasTarget () const
	Returns: true iff an error target has been set
ErrorTarget &	getErrorTarget () const
	Returns: the ErrorTarget for this source
SharedErrorTarget	getSharedErrorTarget () const
	Returns: the SharedErrorTarget for this source
std::string	getErrorSourceName () const
	Gets the name of this source.
void	setErrorSourceName (std::string const &n)
	Sets the name of this source.
void	disableTarget ()
Protected Member Functions
virtual AggComputer *	newAggComputer (AggFunction aggFunction, TupleAttributeDescriptor const *pAttrDesc)
ExecStreamResult	precheckConduitBuffers ()
	Checks the state of the input and output buffers.
virtual void	closeImpl ()
	Implements ClosableObject.
Protected Attributes
SharedExecStreamBufAccessor	pInAccessor
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
SharedExecStreamBufAccessor	pOutAccessor
Private Types
enum	State { STATE_ACCUMULATING, STATE_PRODUCING, STATE_DONE }
Private Member Functions
void	clearAccumulator ()
void	updateAccumulator ()
void	computeOutput ()
void	copyPrevGroupByKey ()
void	setCurGroupByKey ()
int	compareGroupByKeys ()
ExecStreamResult	produce ()
Private Attributes
State	state
AggComputerList	aggComputers
int	groupByKeyCount
TupleData	inputTuple
TupleDataWithBuffer	prevTuple
TupleData	outputTuple
bool	prevTupleValid