dbcon/joblist/windowfunctionstep.cpp

/* Copyright (C) 2014 InfiniDB, Inc.
   Copyright (c) 2016-2020 MariaDB

   This program is free software; you can redistribute it and/or
   modify it under the terms of the GNU General Public License
   as published by the Free Software Foundation; version 2 of
   the License.

   This program is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty of
   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
   GNU General Public License for more details.

   You should have received a copy of the GNU General Public License
   along with this program; if not, write to the Free Software
   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston,
   MA 02110-1301, USA. */

//  $Id: windowfunctionstep.cpp 9681 2013-07-11 22:58:05Z xlou $


//#define NDEBUG
#include <cassert>
#include <sstream>
#include <iomanip>
using namespace std;

#include <boost/algorithm/string.hpp>  //  to_upper_copy
#include <boost/shared_ptr.hpp>
#include <boost/shared_array.hpp>
#include <boost/thread.hpp>
#include <boost/uuid/uuid_io.hpp>
using namespace boost;

#include "atomicops.h"
using namespace atomicops;

#include "loggingid.h"
#include "errorcodes.h"
#include "idberrorinfo.h"
using namespace logging;

#include "configcpp.h"
using namespace config;

#include "calpontselectexecutionplan.h"
#include "calpontsystemcatalog.h"
#include "aggregatecolumn.h"
#include "arithmeticcolumn.h"
#include "constantcolumn.h"
#include "functioncolumn.h"
#include "pseudocolumn.h"
#include "simplefilter.h"
#include "windowfunctioncolumn.h"
using namespace execplan;

#include "../../utils/windowfunction/windowfunction.h"
#include "../../utils/windowfunction/windowfunctiontype.h"
#include "../../utils/windowfunction/framebound.h"
#include "../../utils/windowfunction/frameboundrange.h"
#include "../../utils/windowfunction/frameboundrow.h"
#include "../../utils/windowfunction/windowframe.h"
using namespace windowfunction;

#include "rowgroup.h"
using namespace rowgroup;

using namespace ordering;

#include "funcexp.h"
using namespace funcexp;

#include "querytele.h"
using namespace querytele;

#include "jlf_common.h"
#include "jobstep.h"
#include "windowfunctionstep.h"
using namespace joblist;

#include "checks.h"


namespace
{


uint64_t getColumnIndex(const SRCP& c, const map<uint64_t, uint64_t>& m, JobInfo& jobInfo)
{
    uint64_t key = getTupleKey(jobInfo, c, true);
    const SimpleColumn* sc = dynamic_cast<const SimpleColumn*>(c.get());

    if (sc != NULL && !sc->schemaName().empty())
    {
        // special handling for dictionary
        CalpontSystemCatalog::ColType ct = sc->colType();

//XXX use this before connector sets colType in sc correctly.
//    type of pseudo column is set by connector
        if (!(dynamic_cast<const PseudoColumn*>(sc)))
        {
            ct = jobInfo.csc->colType(sc->oid());
            ct.charsetNumber =sc->colType().charsetNumber;
        }

//X
        CalpontSystemCatalog::OID dictOid = isDictCol(ct);
        string alias(extractTableAlias(sc));

        if (dictOid > 0)
        {
            TupleInfo ti =
                setTupleInfo(ct, dictOid, jobInfo, tableOid(sc, jobInfo.csc), sc, alias);
            key = ti.key;
        }
    }

    map<uint64_t, uint64_t>::const_iterator j = m.find(key);

    if (j == m.end())
    {
        string name = jobInfo.keyInfo->tupleKeyToName[key];
        cerr << name << " is not in tuple, key=" << key << endl;
        throw IDBExcept(IDBErrorInfo::instance()->errorMsg(ERR_WF_COLUMN_MISSING, name),
                        ERR_WF_COLUMN_MISSING);
    }

    return j->second;
}


}


namespace joblist
{


WindowFunctionStep::WindowFunctionStep(const JobInfo& jobInfo) :
    JobStep(jobInfo),
    fRunner(0),
    fCatalog(jobInfo.csc),
    fRowsReturned(0),
    fEndOfResult(false),
    fIsSelect(true),
    fUseSSMutex(false),
    fUseUFMutex(false),
    fInputDL(NULL),
    fOutputDL(NULL),
    fInputIterator(-1),
    fOutputIterator(-1),
    fFunctionCount(0),
    fTotalThreads(1),
    fNextIndex(0),
    fMemUsage(0),
    fRm(jobInfo.rm),
    fSessionMemLimit(jobInfo.umMemLimit)
{
    fTotalThreads = fRm->windowFunctionThreads();
    fExtendedInfo = "WFS: ";
    fQtc.stepParms().stepType = StepTeleStats::T_WFS;
}


WindowFunctionStep::~WindowFunctionStep()
{
    if (fMemUsage > 0)
        fRm->returnMemory(fMemUsage, fSessionMemLimit);
}


void WindowFunctionStep::run()
{
    if (fInputJobStepAssociation.outSize() == 0)
        throw logic_error("No input data list for window function step.");

    fInputDL = fInputJobStepAssociation.outAt(0)->rowGroupDL();

    if (fInputDL == NULL)
        throw logic_error("Input is not a RowGroup data list in window function step.");

    fInputIterator = fInputDL->getIterator();

    if (fOutputJobStepAssociation.outSize() == 0)
        throw logic_error("No output data list for window function step.");

    fOutputDL = fOutputJobStepAssociation.outAt(0)->rowGroupDL();

    if (fOutputDL == NULL)
        throw logic_error("Output of window function step is not a RowGroup data list.");

    if (fDelivery == true)
    {
        fOutputIterator = fOutputDL->getIterator();
    }

    fRunner = jobstepThreadPool.invoke(Runner(this));
}


void WindowFunctionStep::join()
{
    if (fRunner)
        jobstepThreadPool.join(fRunner);
}


uint32_t WindowFunctionStep::nextBand(messageqcpp::ByteStream& bs)
{
    RGData rgDataOut;
    bool more = false;
    uint32_t rowCount = 0;

    try
    {
        bs.restart();

        more = fOutputDL->next(fOutputIterator, &rgDataOut);

        if (more && !cancelled())
        {
            fRowGroupDelivered.setData(&rgDataOut);
            fRowGroupDelivered.serializeRGData(bs);
            rowCount = fRowGroupDelivered.getRowCount();
        }
        else
        {
            while (more)
                more = fOutputDL->next(fOutputIterator, &rgDataOut);

            fEndOfResult = true;
        }
    }
    catch (...)
    {
        handleException(std::current_exception(),
                        logging::ERR_IN_DELIVERY,
                        logging::ERR_WF_DATA_SET_TOO_BIG,
                        "WindowFunctionStep::nextBand()");
        while (more)
            more = fOutputDL->next(fOutputIterator, &rgDataOut);
        fEndOfResult = true;
    }

    if (fEndOfResult)
    {
        // send an empty / error band
        rgDataOut.reinit(fRowGroupDelivered, 0);
        fRowGroupDelivered.setData(&rgDataOut);
        fRowGroupDelivered.resetRowGroup(0);
        fRowGroupDelivered.setStatus(status());
        fRowGroupDelivered.serializeRGData(bs);
    }

    return rowCount;
}


void WindowFunctionStep::setOutputRowGroup(const RowGroup& rg)
{
    idbassert(0);
}


const RowGroup& WindowFunctionStep::getOutputRowGroup() const
{
    return fRowGroupOut;
}


const RowGroup& WindowFunctionStep::getDeliveredRowGroup() const
{
    return fRowGroupDelivered;
}


void WindowFunctionStep::deliverStringTableRowGroup(bool b)
{
    fRowGroupOut.setUseStringTable(b);
    fRowGroupDelivered.setUseStringTable(b);
}


bool WindowFunctionStep::deliverStringTableRowGroup() const
{
    idbassert(fRowGroupOut.usesStringTable() == fRowGroupDelivered.usesStringTable());
    return fRowGroupDelivered.usesStringTable();
}


const string WindowFunctionStep::toString() const
{
    ostringstream oss;
    oss << "WindowFunctionStep   ses:" << fSessionId << " txn:" << fTxnId << " st:" << fStepId;

    oss << " in:";

    for (unsigned i = 0; i < fInputJobStepAssociation.outSize(); i++)
        oss << fInputJobStepAssociation.outAt(i);

    if (fOutputJobStepAssociation.outSize() > 0)
    {
        oss << " out:";

        for (unsigned i = 0; i < fOutputJobStepAssociation.outSize(); i++)
            oss << fOutputJobStepAssociation.outAt(i);
    }

    return oss.str();
}

void WindowFunctionStep::AddSimplColumn(const vector<SimpleColumn*>& scs,
                                     JobInfo& jobInfo)
{
    // append the simple columns if not already projected
    set<UniqId> scProjected;

    for (RetColsVector::iterator i  = jobInfo.projectionCols.begin();
            i != jobInfo.projectionCols.end();
            i++)
    {
        SimpleColumn* sc = dynamic_cast<SimpleColumn*>(i->get());

        if (sc != NULL)
        {
            if (sc->schemaName().empty())
                sc->oid(joblist::tableOid(sc, jobInfo.csc) + 1 + sc->colPosition());

            scProjected.insert(UniqId(sc));
        }
    }

    for (vector<SimpleColumn*>::const_iterator i = scs.begin(); i != scs.end(); i++)
    {
        if (scProjected.find(UniqId(*i)) == scProjected.end())
        {
            jobInfo.windowDels.push_back(SRCP((*i)->clone()));
// MCOL-3343 Enable this if we decide to allow Window Functions to run with
// aggregates with no group by. MariaDB allows this. Nobody else in the world does.
// There will be more work to get it to function if we try this.
//            jobInfo.windowSet.insert(getTupleKey(jobInfo, *i, true));
            scProjected.insert(UniqId(*i));
        }
    }
}

void WindowFunctionStep::checkWindowFunction(CalpontSelectExecutionPlan* csep, JobInfo& jobInfo)
{
    // window functions in select clause, selected or in expression
    jobInfo.windowDels = jobInfo.deliveredCols;

    for (RetColsVector::iterator i = jobInfo.windowDels.begin(); i < jobInfo.windowDels.end(); i++)
    {
        const vector<WindowFunctionColumn*>& wcl = (*i)->windowfunctionColumnList();
        RetColsVector wcList;

        for (vector<WindowFunctionColumn*>::const_iterator j = wcl.begin(); j != wcl.end(); j++)
            wcList.push_back(SRCP((*j)->clone()));

        if (!wcList.empty())
        {
            jobInfo.windowExps.push_back(*i);
            jobInfo.windowSet.insert(getTupleKey(jobInfo, *i, true));
        }

        if (dynamic_cast<WindowFunctionColumn*>(i->get()) != NULL)
        {
            jobInfo.windowCols.push_back(*i);
            jobInfo.windowSet.insert(getTupleKey(jobInfo, *i, true));
        }
        else if (!wcList.empty())
        {
            jobInfo.windowCols.insert(jobInfo.windowCols.end(), wcList.begin(), wcList.end());

            for (RetColsVector::const_iterator k = wcList.begin(); k < wcList.end(); k++)
            {
                jobInfo.windowSet.insert(getTupleKey(jobInfo, *k, true));
            }
        }
    }

    // window functions in order by clause
    const CalpontSelectExecutionPlan::OrderByColumnList& orderByCols = csep->orderByCols();
    RetColsVector wcInOrderby;

    for (uint64_t i = 0; i < orderByCols.size(); i++)
    {
        if (orderByCols[i]->orderPos() == (uint64_t)(-1))
        {
            WindowFunctionColumn* wc = dynamic_cast<WindowFunctionColumn*>(orderByCols[i].get());
            const vector<WindowFunctionColumn*>& wcl = orderByCols[i]->windowfunctionColumnList();
            RetColsVector wcList;

            for (vector<WindowFunctionColumn*>::const_iterator j = wcl.begin(); j != wcl.end(); j++)
                wcList.push_back(SRCP((*j)->clone()));

            if (wc == NULL && wcList.empty())
                continue;

            // an window function or expression of window functions
            wcInOrderby.push_back(orderByCols[i]);

            if (!wcList.empty())
            {
                jobInfo.windowExps.push_back(orderByCols[i]);
                jobInfo.windowSet.insert(getTupleKey(jobInfo, orderByCols[i], true));
            }

            if (dynamic_cast<WindowFunctionColumn*>(orderByCols[i].get()) != NULL)
            {
                jobInfo.windowCols.push_back(orderByCols[i]);
                jobInfo.windowSet.insert(getTupleKey(jobInfo, orderByCols[i], true));
            }
            else if (!wcList.empty())
            {
                jobInfo.windowCols.insert(
                    jobInfo.windowCols.end(), wcList.begin(), wcList.end());

                for (RetColsVector::const_iterator k = wcList.begin(); k < wcList.end(); k++)
                {
                    jobInfo.windowSet.insert(getTupleKey(jobInfo, *k, true));
                }
            }
        }
    }

    // no window function involved in the query
    if (jobInfo.windowCols.empty())
        return;

    // Add in the non-window side of arithmetic columns and functions
    for (uint64_t i = 0; i < jobInfo.windowExps.size(); i++)
    {
        const ArithmeticColumn* ac =
            dynamic_cast<const ArithmeticColumn*>(jobInfo.windowExps[i].get());
        const FunctionColumn* fc =
            dynamic_cast<const FunctionColumn*>(jobInfo.windowExps[i].get());

        if (ac != NULL && ac->windowfunctionColumnList().size() > 0)
        {
            AddSimplColumn(ac->simpleColumnList(), jobInfo);
        }
        else if (fc != NULL && fc->windowfunctionColumnList().size() > 0)
        {
            AddSimplColumn(fc->simpleColumnList(), jobInfo);
        }
    }
    // reconstruct the delivered column list with auxiliary columns
    set<uint64_t> colSet;
    jobInfo.deliveredCols.resize(0);

    for (RetColsVector::iterator i = jobInfo.windowDels.begin(); i < jobInfo.windowDels.end(); i++)
    {
        jobInfo.deliveredCols.push_back(*i);
        uint64_t key = getTupleKey(jobInfo, *i, true);

        // TODO: remove duplicates in select clause
        colSet.insert(key);
    }

    // add window columns in orderby
    for (RetColsVector::iterator i = wcInOrderby.begin(); i < wcInOrderby.end(); i++)
    {
        jobInfo.deliveredCols.push_back(*i);
        uint64_t key = getTupleKey(jobInfo, *i, true);
        colSet.insert(key);
    }

    // MCOL-3435 We haven't yet checked for aggregate, but we need to know
    bool hasAggregation = false;
    for (uint64_t i = 0; i < jobInfo.deliveredCols.size(); i++)
    {
        if (dynamic_cast<AggregateColumn*>(jobInfo.deliveredCols[i].get()) != NULL)
        {
            hasAggregation = true;
            break;
        }
    }

    // add non-duplicate auxiliary columns
    RetColsVector colsInAf;

    for (RetColsVector::iterator i = jobInfo.windowCols.begin(); i < jobInfo.windowCols.end(); i++)
    {
        uint64_t key = getTupleKey(jobInfo, *i, true);

        if (colSet.find(key) == colSet.end())
            jobInfo.deliveredCols.push_back(*i);

        RetColsVector columns = dynamic_cast<WindowFunctionColumn*>(i->get())->getColumnList();

        for (RetColsVector::iterator j = columns.begin(); j < columns.end(); j++)
        {
            if (dynamic_cast<ConstantColumn*>(j->get()) != NULL)
                continue;

            key = getTupleKey(jobInfo, *j, true);

            if (colSet.find(key) == colSet.end())
            {
                jobInfo.deliveredCols.push_back(*j);
                // MCOL-3435 Allow Window Functions to run with aggregates with
                // no group by by inserting a group by for window parameters.
                if (hasAggregation)
                {
                    // If an argument is an AggregateColumn, don't group by it.
                    if (dynamic_cast<AggregateColumn*>(j->get()) == NULL)
                    {
                        bool bFound = false;
                        for (std::vector<SRCP>::iterator igpc = csep->groupByCols().begin();
                                                         igpc < csep->groupByCols().end();
                                                         ++igpc)
                        {
                            if ((*igpc)->alias() == (*j)->alias())
                            {
                                bFound = true;
                                break;
                            }
                        }
                        if (!bFound)
                        {
                            csep->groupByCols().push_back(*j);
                        }
                    }
                }
            }

            colSet.insert(key);
        }
    }

    // for handling order by and limit in outer query
    jobInfo.wfqLimitStart = csep->limitStart();
    jobInfo.wfqLimitCount = csep->limitNum();
    csep->limitStart(0);
    csep->limitNum(-1);

    if (csep->orderByCols().size() > 0)
    {
        jobInfo.wfqOrderby = csep->orderByCols();
        csep->orderByCols().clear();

        // add order by columns
        for (RetColsVector::iterator i = jobInfo.wfqOrderby.begin(); i < jobInfo.wfqOrderby.end(); i++)
        {
            if (dynamic_cast<ConstantColumn*>(i->get()) != NULL)
                continue;

            uint64_t key = getTupleKey(jobInfo, *i, true);

            if (colSet.find(key) == colSet.end())
                jobInfo.deliveredCols.push_back(*i);

            colSet.insert(key);
        }
    }
}


SJSTEP WindowFunctionStep::makeWindowFunctionStep(SJSTEP& step, JobInfo& jobInfo)
{
    // create a window function step
    WindowFunctionStep* ws = new WindowFunctionStep(jobInfo);

    // connect to the feeding step
    JobStepAssociation jsa;
    AnyDataListSPtr spdl(new AnyDataList());
    RowGroupDL* dl = new RowGroupDL(1, jobInfo.fifoSize);
    dl->OID(execplan::CNX_VTABLE_ID);
    spdl->rowGroupDL(dl);
    jsa.outAdd(spdl);
    ws->inputAssociation(jsa);
    ws->stepId(step->stepId() + 1);
    step->outputAssociation(jsa);

    AnyDataListSPtr spdlOut(new AnyDataList());
    RowGroupDL* dlOut = new RowGroupDL(1, jobInfo.fifoSize);
    dlOut->OID(CNX_VTABLE_ID);
    spdlOut->rowGroupDL(dlOut);
    JobStepAssociation jsaOut;
    jsaOut.outAdd(spdlOut);
    ws->outputAssociation(jsaOut);

    // configure the rowgroups and index mapping
    TupleDeliveryStep* ds = dynamic_cast<TupleDeliveryStep*>(step.get());
    idbassert(ds != NULL);
    ws->initialize(ds->getDeliveredRowGroup(), jobInfo);

    // restore the original delivery coloumns
    jobInfo.deliveredCols = jobInfo.windowDels;
    jobInfo.nonConstDelCols.clear();

    for (RetColsVector::iterator i = jobInfo.windowDels.begin(); i < jobInfo.windowDels.end(); i++)
    {
        if (NULL == dynamic_cast<const ConstantColumn*>(i->get()))
            jobInfo.nonConstDelCols.push_back(*i);
    }

    return SJSTEP(ws);
}


void WindowFunctionStep::initialize(const RowGroup& rg, JobInfo& jobInfo)
{
    if (jobInfo.trace) cout << "Input to WindowFunctionStep: " << rg.toString() << endl;

    // query type decides the output by dbroot or partition
    // @bug 5631. Insert select should be treated as select
    fIsSelect = (jobInfo.queryType == "SELECT" ||
                 jobInfo.queryType == "INSERT_SELECT");

    // input row meta data
    fRowGroupIn = rg;
    fRowGroupIn.initRow(&fRowIn);

    // make an input map(id, index)
    map<uint64_t, uint64_t> colIndexMap;
    uint64_t colCntIn = rg.getColumnCount();
    const vector<uint32_t>& pos = rg.getOffsets();
    const vector<uint32_t>& oids = rg.getOIDs();
    const vector<uint32_t>& keys = rg.getKeys();
    const vector<CalpontSystemCatalog::ColDataType>& types = rg.getColTypes();
    const vector<uint32_t>& csNums = rg.getCharsetNumbers();
    const vector<uint32_t>& scales = rg.getScale();
    const vector<uint32_t>& precisions = rg.getPrecision();

    for (uint64_t i = 0; i < colCntIn; i++)
        colIndexMap.insert(make_pair(keys[i], i));

    // @bug6065, window functions that will update string table
    int64_t wfsUpdateStringTable = 0;
    int64_t wfsUserFunctionCount = 0;

    for (RetColsVector::iterator i = jobInfo.windowCols.begin(); i < jobInfo.windowCols.end(); i++)
    {
        bool isUDAF = false;
        // window function type
        WindowFunctionColumn* wc = dynamic_cast<WindowFunctionColumn*>(i->get());
        uint64_t ridx = getColumnIndex(*i, colIndexMap, jobInfo);    // result index
        // @bug6065, window functions that will update string table
        {
            CalpontSystemCatalog::ColType rt = wc->resultType();

            if ((types[ridx] == CalpontSystemCatalog::CHAR ||
                    types[ridx] == CalpontSystemCatalog::VARCHAR ||
                    types[ridx] == CalpontSystemCatalog::TEXT ||
                    types[ridx] == CalpontSystemCatalog::VARBINARY ||
                    types[ridx] == CalpontSystemCatalog::BLOB) &&
                    rg.getColumnWidth(ridx) >= jobInfo.stringTableThreshold)
            {
                ++wfsUpdateStringTable;
            }
        }

//		if (boost::iequals(wc->functionName(),"UDAF_FUNC")
        if (wc->functionName() == "UDAF_FUNC")
        {
            isUDAF = true;
            ++wfsUserFunctionCount;
        }

        vector<int64_t> fields;
        fields.push_back(ridx);  // result
        const RetColsVector& parms = wc->functionParms();

        for (uint64_t i = 0; i < parms.size(); i++)                  // arguments
        {
            // skip constant column
            if (dynamic_cast<const ConstantColumn*>(parms[i].get()) == NULL)
                fields.push_back(getColumnIndex(parms[i], colIndexMap, jobInfo));
            else
                fields.push_back(-1);
        }

        // partition & order by
        const RetColsVector& partitions = wc->partitions();
        vector<uint64_t> eqIdx;
        vector<uint64_t> peerIdx;
        vector<IdbSortSpec> sorts;

        for (uint64_t i = 0; i < partitions.size(); i++)
        {
            // skip constant column
            if (dynamic_cast<const ConstantColumn*>(partitions[i].get()) != NULL)
                continue;

            // get column index
            uint64_t idx = getColumnIndex(partitions[i], colIndexMap, jobInfo);
            eqIdx.push_back(idx);
            sorts.push_back(IdbSortSpec(idx, partitions[i]->asc(), partitions[i]->nullsFirst()));
        }

        const RetColsVector& orders = wc->orderBy().fOrders;

        for (uint64_t i = 0; i < orders.size(); i++)
        {
            // skip constant column
            if (dynamic_cast<const ConstantColumn*>(orders[i].get()) != NULL)
                continue;

            // get column index
            uint64_t idx = getColumnIndex(orders[i], colIndexMap, jobInfo);
            peerIdx.push_back(idx);
            sorts.push_back(IdbSortSpec(idx, orders[i]->asc(), orders[i]->nullsFirst()));
        }

        // functors for sorting
        boost::shared_ptr<EqualCompData> parts(new EqualCompData(eqIdx, rg));
        boost::shared_ptr<OrderByData> orderbys(new OrderByData(sorts, rg));
        boost::shared_ptr<EqualCompData> peers(new EqualCompData(peerIdx, rg));

        // column type for functor templates
        int ct = 0;

        if (isUDAF)
        {
            ct = wc->getUDAFContext().getResultType();
        }
        // make sure index is in range
        else if (fields.size() > 1 && fields[1] >= 0 && static_cast<uint64_t>(fields[1]) < types.size())
            ct = types[fields[1]];

        // workaround for functions using "within group (order by)" syntax
        string fn = boost::to_upper_copy(wc->functionName());

        if ( (fn == "MEDIAN" || fn == "PERCENTILE_CONT" || fn == "PERCENTILE_DISC") &&
                utils::is_nonnegative(peerIdx[0]) && peerIdx[0] < types.size() )
            ct = types[peerIdx[0]];

        // create the functor based on function name
        boost::shared_ptr<WindowFunctionType> func =
            WindowFunctionType::makeWindowFunction(fn, ct, wc);

        // parse parms after peer and fields are set
        // functions may need to set order column index
        func->peer(peers);
        func->fieldIndex(fields);
        func->parseParms(parms);

        // window frame
        const WF_Frame& frame = wc->orderBy().fFrame;
        int frameUnit = (frame.fIsRange) ? WF__FRAME_RANGE : WF__FRAME_ROWS;

        if (frame.fStart.fFrame == WF_UNBOUNDED_PRECEDING &&
                frame.fEnd.fFrame == WF_UNBOUNDED_FOLLOWING)
            frameUnit = WF__FRAME_ROWS;

        boost::shared_ptr<FrameBound> upper = parseFrameBound(
                frame.fStart, colIndexMap, orders, peers, jobInfo, !frame.fIsRange, true);
        boost::shared_ptr<FrameBound> lower = parseFrameBound(
                frame.fEnd, colIndexMap, orders, peers, jobInfo, !frame.fIsRange, false);
        boost::shared_ptr<WindowFrame> windows(new WindowFrame(frameUnit, upper, lower));
        func->frameUnit(frameUnit);

        // add to the function list
        fFunctions.push_back(boost::shared_ptr<WindowFunction>(
                                 new WindowFunction(func, parts, orderbys, windows, rg, fRowIn)));
        fFunctionCount++;
    }

    // initialize window function expresssions
    fExpression = jobInfo.windowExps;

    for (RetColsVector::iterator i = fExpression.begin(); i < fExpression.end(); i++)
    {
        // output index
        (*i)->outputIndex(getColumnIndex(*i, colIndexMap, jobInfo));

        // map the input indices
        const vector<SimpleColumn*>& scols = (*i)->simpleColumnList();

        for (vector<SimpleColumn*>::const_iterator j = scols.begin(); j != scols.end(); j++)
        {
            uint64_t key = getTupleKey(jobInfo, *j);
            CalpontSystemCatalog::OID dictOid = joblist::isDictCol((*j)->colType());

            if (dictOid > 0)
            {
                key = jobInfo.keyInfo->dictKeyMap[key];
            }

            map<uint64_t, uint64_t>::iterator k = colIndexMap.find(key);

            if (k != colIndexMap.end())
            {
                (*j)->inputIndex(k->second);
            }
            else
            {
                string name = jobInfo.keyInfo->tupleKeyToName[key];
                cerr << name << " is not in tuple, key=" << key << endl;
                throw IDBExcept(IDBErrorInfo::instance()->errorMsg(ERR_WF_COLUMN_MISSING, name),
                                ERR_WF_COLUMN_MISSING);
            }
        }

        ArithmeticColumn* ac = dynamic_cast<ArithmeticColumn*>((*i).get());
        FunctionColumn*   fc = dynamic_cast<FunctionColumn*>((*i).get());

        if (ac != NULL)
        {
            updateWindowCols(ac->expression(), colIndexMap, jobInfo);
        }
        else if (fc != NULL)
        {
//			RetColsVector wcList = fc->windowfunctionColumnList();
//			for (RetColsVector::iterator j = wcList.begin(); j != wcList.end(); j++)
//				(*j)->inputIndex(getColumnIndex(*j, colIndexMap, jobInfo));
            funcexp::FunctionParm parms = fc->functionParms();

            for (vector<execplan::SPTP>::iterator j = parms.begin(); j < parms.end(); j++)
                updateWindowCols(j->get(), colIndexMap, jobInfo);
        }
    }

    // order by part
    if (jobInfo.wfqOrderby.size() > 0)
    {
        // query order by
        vector<uint64_t> eqIdx;
        vector<IdbSortSpec> sorts;
        const RetColsVector& orderby = jobInfo.wfqOrderby;

        for (uint64_t i = 0; i < orderby.size(); i++)
        {
            // skip constant column
            if (dynamic_cast<const ConstantColumn*>(orderby[i].get()) != NULL)
                continue;

            // get column index
            uint64_t idx = getColumnIndex(orderby[i], colIndexMap, jobInfo);
            sorts.push_back(IdbSortSpec(idx, orderby[i]->asc(), orderby[i]->nullsFirst()));
        }

        fQueryOrderBy.reset(new OrderByData(sorts, rg));
    }

    // limit part
    fQueryLimitStart = jobInfo.wfqLimitStart;
    fQueryLimitCount = jobInfo.wfqLimitCount;

    // fix the delivered rowgroup data
    vector<uint64_t> delColIdx;

    for (RetColsVector::iterator i = jobInfo.windowDels.begin(); i < jobInfo.windowDels.end(); i++)
    {
        // find the none constantant columns in the deliver
        // leave constants to annexstep for now.
        if (dynamic_cast<const ConstantColumn*>((*i).get()) != NULL)
            continue;

        delColIdx.push_back(getColumnIndex(*i, colIndexMap, jobInfo));
    }

    size_t retColCount = delColIdx.size();
    vector<uint32_t> pos1;
    vector<uint32_t> oids1;
    vector<uint32_t> keys1;
    vector<uint32_t> scales1;
    vector<uint32_t> precisions1;
    vector<CalpontSystemCatalog::ColDataType> types1;
    vector<uint32_t> csNums1;
    pos1.push_back(2);

    for (size_t i = 0; i < retColCount; i++)
    {
        size_t j = delColIdx[i];
        pos1.push_back(pos1[i] + (pos[j + 1] - pos[j]));
        oids1.push_back(oids[j]);
        keys1.push_back(keys[j]);
        scales1.push_back(scales[j]);
        precisions1.push_back(precisions[j]);
        types1.push_back(types[j]);
        csNums1.push_back(csNums[j]);
    }

    fRowGroupDelivered = RowGroup(
                             retColCount, pos1, oids1, keys1, types1, csNums1, scales1, precisions1, jobInfo.stringTableThreshold);

    if (jobInfo.trace)
        cout << "delivered RG: " << fRowGroupDelivered.toString() << endl << endl;

    if (wfsUpdateStringTable > 1)
        fUseSSMutex = true;

    if (wfsUserFunctionCount > 1)
        fUseUFMutex = true;

    fRowGroupOut = fRowGroupDelivered;
}


void WindowFunctionStep::execute()
{
    RGData rgData;
    Row row;
    fRowGroupIn.initRow(&row);
    bool more = fInputDL->next(fInputIterator, &rgData);
    uint64_t i = 0; // for RowGroup index in the fInRowGroupData

    if (traceOn()) dlTimes.setFirstReadTime();

    StepTeleStats sts;
    sts.query_uuid = fQueryUuid;
    sts.step_uuid = fStepUuid;
    sts.msg_type = StepTeleStats::ST_START;
    sts.total_units_of_work = 1;
    postStepStartTele(sts);

    try
    {
        while (more && !cancelled())
        {
            fRowGroupIn.setData(&rgData);
            fRowGroupIn.getRow(0, &row);
            uint64_t rowCnt = fRowGroupIn.getRowCount();

            if (rowCnt > 0)
            {
                fInRowGroupData.push_back(rgData);
                uint64_t memAdd = fRowGroupIn.getSizeWithStrings() + rowCnt * sizeof(RowPosition);
                fMemUsage += memAdd;

                if (fRm->getMemory(memAdd, fSessionMemLimit) == false)
                    throw IDBExcept(ERR_WF_DATA_SET_TOO_BIG);

                for (uint64_t j = 0; j < rowCnt; ++j)
                {
                    if (i > 0x0000FFFFFFFFFFFFULL || j > 0x000000000000FFFFULL)
                        throw IDBExcept(ERR_WF_DATA_SET_TOO_BIG);

                    fRows.push_back(RowPosition(i, j));
                    row.nextRow();
                }

                //@bug6065, make StringStore::storeString() thread safe, default to false.
                rgData.useStoreStringMutex(fUseSSMutex);
                // For the User Data of UDAnF
                rgData.useUserDataMutex(fUseUFMutex);

                // window function does not change row count
                fRowsReturned += rowCnt;

                i++;
            }

            more = fInputDL->next(fInputIterator, &rgData);
        }
    } // try
    catch (...)
    {
        handleException(std::current_exception(),
                        logging::ERR_READ_INPUT_DATALIST,
                        logging::ERR_WF_DATA_SET_TOO_BIG,
                        "WindowFunctionStep::execute()");
    }

    if (traceOn())
        dlTimes.setLastReadTime();

    // no need for the window function if aborted or result set is empty.
    if (cancelled() || fRows.size() == 0)
    {
        while (more)
            more = fInputDL->next(fInputIterator, &rgData);

        fOutputDL->endOfInput();

        sts.msg_type = StepTeleStats::ST_SUMMARY;
        sts.total_units_of_work = sts.units_of_work_completed = 1;
        sts.rows = fRowsReturned;
        postStepSummaryTele(sts);

        if (traceOn())
        {
            dlTimes.setEndOfInputTime();
            printCalTrace();
        }

        return;
    }

    // got something to work on
    try
    {
        if (fFunctionCount == 1)
        {
            doFunction();
        }
        else
        {
            if (fTotalThreads > fFunctionCount)
                fTotalThreads = fFunctionCount;

            fFunctionThreads.clear();
            fFunctionThreads.reserve(fTotalThreads);

            for (uint64_t i = 0; i < fTotalThreads && !cancelled(); i++)
                fFunctionThreads.push_back(jobstepThreadPool.invoke(WFunction(this)));

            // If cancelled, not all threads are started.
            jobstepThreadPool.join(fFunctionThreads);
        }

        if (!(cancelled()))
        {
            if (fIsSelect)
                doPostProcessForSelect();
            else
                doPostProcessForDml();
        }

    }
    catch (...)
    {
        handleException(std::current_exception(),
                        logging::ERR_EXECUTE_WINDOW_FUNCTION,
                        logging::ERR_WF_DATA_SET_TOO_BIG,
                        "WindowFunctionStep::execute()");
    }

    fOutputDL->endOfInput();

    sts.msg_type = StepTeleStats::ST_SUMMARY;
    sts.total_units_of_work = sts.units_of_work_completed = 1;
    sts.rows = fRowsReturned;
    postStepSummaryTele(sts);

    if (traceOn())
    {
        dlTimes.setEndOfInputTime();
        printCalTrace();
    }

    return;
}


uint64_t WindowFunctionStep::nextFunctionIndex()
{
    uint64_t idx = atomicInc(&fNextIndex);

    // return index in the function array
    return --idx;
}


void WindowFunctionStep::doFunction()
{
    uint64_t i = 0;

    try
    {
        while (((i = nextFunctionIndex()) < fFunctionCount) && !cancelled())
        {
            uint64_t memAdd = fRows.size() * sizeof(RowPosition);
            fMemUsage += memAdd;

            if (fRm->getMemory(memAdd, fSessionMemLimit) == false)
                throw IDBExcept(ERR_WF_DATA_SET_TOO_BIG);

            fFunctions[i]->setCallback(this, i);
            (*fFunctions[i].get())();
        }
    }
    catch (...)
    {
        handleException(std::current_exception(),
                        logging::ERR_EXECUTE_WINDOW_FUNCTION,
                        logging::ERR_WF_DATA_SET_TOO_BIG,
                        "WindowFunctionStep::doFunction()");
    }
}

void WindowFunctionStep::doPostProcessForSelect()
{
    FuncExp* fe = funcexp::FuncExp::instance();
    boost::shared_array<int> mapping = makeMapping(fRowGroupIn, fRowGroupOut);
    Row rowIn, rowOut;
    fRowGroupIn.initRow(&rowIn);
    fRowGroupOut.initRow(&rowOut);
    RGData rgData;
    vector<RowPosition>& rowData = *(fFunctions.back()->fRowData.get());
    int64_t rowsLeft = rowData.size();
    int64_t rowsInRg = 0;
    int64_t rgCapacity = 0;

    int64_t begin = fQueryLimitStart;
    int64_t count = (fQueryLimitCount == (uint64_t) - 1) ? rowsLeft : fQueryLimitCount;
    int64_t end = begin + count;
    end = (end < rowsLeft) ? end : rowsLeft;
    rowsLeft = (end > begin) ? (end - begin) : 0;

    if (fQueryOrderBy.get() != NULL)
        sort(rowData.begin(), rowData.size());

    for (int64_t i = begin; i < end; i++)
    {
        if (rgData.rowData.get() == NULL)
        {
            rgCapacity = ((rowsLeft > 8192) ? 8192 : rowsLeft);
            rowsLeft -= rgCapacity;
            rgData.reinit(fRowGroupOut, rgCapacity);

            fRowGroupOut.setData(&rgData);
            fRowGroupOut.resetRowGroup(0);
            fRowGroupOut.setDBRoot(0);           // not valid dbroot
            fRowGroupOut.getRow(0, &rowOut);
            rowsInRg = 0;
        }

        rowIn.setData(getPointer(rowData[i], fRowGroupIn, rowIn));

        // evaluate the window function expressions before apply mapping
        if (fExpression.size() > 0)
            fe->evaluate(rowIn, fExpression);

        applyMapping(mapping, rowIn, &rowOut);
        rowOut.nextRow();
        rowsInRg++;

        if (rowsInRg == rgCapacity)
        {
            fRowGroupOut.setRowCount(rowsInRg);
            fOutputDL->insert(rgData);
            rgData.clear();
        }
    }
}


void WindowFunctionStep::doPostProcessForDml()
{
    FuncExp* fe = funcexp::FuncExp::instance();
    boost::shared_array<int> mapping = makeMapping(fRowGroupIn, fRowGroupOut);
    Row rowIn, rowOut;
    fRowGroupIn.initRow(&rowIn);
    fRowGroupOut.initRow(&rowOut);

    for (vector<RGData>::iterator i = fInRowGroupData.begin();
            i < fInRowGroupData.end(); i++)
    {
        fRowGroupIn.setData(&(*i));
        RGData rgData = RGData(fRowGroupIn, fRowGroupIn.getRowCount());
        fRowGroupOut.setData(&rgData);
        // @bug 5631. reset rowgroup before the data is populated.
        fRowGroupOut.resetRowGroup(fRowGroupIn.getBaseRid());
        fRowGroupOut.setDBRoot(fRowGroupIn.getDBRoot());
        fRowGroupOut.setRowCount(fRowGroupIn.getRowCount());

        fRowGroupIn.getRow(0, &rowIn);
        fRowGroupOut.getRow(0, &rowOut);

        for (uint64_t i = 0; i < fRowGroupIn.getRowCount(); ++i)
        {
            // evaluate the window function expressions before apply mapping
            if (fExpression.size() > 0)
                fe->evaluate(rowIn, fExpression);

            applyMapping(mapping, rowIn, &rowOut);
            rowIn.nextRow();
            rowOut.nextRow();
        }

        //fRowGroupOut.resetRowGroup(fRowGroupIn.getBaseRid());
        //fRowGroupOut.setRowCount(fRowGroupIn.getRowCount());

        fOutputDL->insert(rgData);
    }
}


boost::shared_ptr<FrameBound> WindowFunctionStep::parseFrameBoundRows(
    const execplan::WF_Boundary& b,
    const map<uint64_t, uint64_t>& m,
    JobInfo& jobInfo)
{
    boost::shared_ptr<FrameBound> fb;

    if (b.fFrame == WF_CURRENT_ROW)
    {
        fb.reset(new FrameBoundRow(WF__CURRENT_ROW));
        return fb;
    }

    ConstantColumn* cc = dynamic_cast<ConstantColumn*>(b.fVal.get());

    if (cc != NULL)
    {
        Row dummy;
        bool isNull = false;
        int val = cc->getIntVal(dummy, isNull);

        if (val >= 0 && isNull == false)
        {
            int type = (b.fFrame == WF_PRECEDING) ? WF__CONSTANT_PRECEDING : WF__CONSTANT_FOLLOWING;
            fb.reset(new FrameBoundConstantRow(type, val));
        }
        else
        {
            string str("NULL");

            if (!isNull)
            {
                ostringstream oss;
                oss << val;
                str = oss.str();
            }

            throw IDBExcept(IDBErrorInfo::instance()->errorMsg(ERR_WF_BOUND_OUT_OF_RANGE, str),
                            ERR_WF_BOUND_OUT_OF_RANGE);
        }
    }
    else
    {
        int type = (b.fFrame == WF_PRECEDING) ? WF__EXPRESSION_PRECEDING : WF__EXPRESSION_FOLLOWING;
        uint64_t id = getTupleKey(jobInfo, b.fVal);
        uint64_t idx = getColumnIndex(b.fVal, m, jobInfo);
        TupleInfo ti = getTupleInfo(id, jobInfo);

        switch (ti.dtype)
        {
            case execplan::CalpontSystemCatalog::TINYINT:
            case execplan::CalpontSystemCatalog::SMALLINT:
            case execplan::CalpontSystemCatalog::MEDINT:
            case execplan::CalpontSystemCatalog::INT:
            case execplan::CalpontSystemCatalog::BIGINT:
            case execplan::CalpontSystemCatalog::DECIMAL:
            {
                fb.reset(new FrameBoundExpressionRow<int64_t>(type, id, idx));
                break;
            }

            case execplan::CalpontSystemCatalog::DOUBLE:
            case execplan::CalpontSystemCatalog::UDOUBLE:
            {
                fb.reset(new FrameBoundExpressionRow<double>(type, id, idx));
                break;
            }

            case execplan::CalpontSystemCatalog::FLOAT:
            case execplan::CalpontSystemCatalog::UFLOAT:
            {
                fb.reset(new FrameBoundExpressionRow<float>(type, id, idx));
                break;
            }

            case execplan::CalpontSystemCatalog::UTINYINT:
            case execplan::CalpontSystemCatalog::USMALLINT:
            case execplan::CalpontSystemCatalog::UMEDINT:
            case execplan::CalpontSystemCatalog::UINT:
            case execplan::CalpontSystemCatalog::UBIGINT:
            case execplan::CalpontSystemCatalog::UDECIMAL:
            case execplan::CalpontSystemCatalog::DATE:
            case execplan::CalpontSystemCatalog::DATETIME:
            case execplan::CalpontSystemCatalog::TIME:
            case execplan::CalpontSystemCatalog::TIMESTAMP:
            {
                fb.reset(new FrameBoundExpressionRow<uint64_t>(type, id, idx));
                break;
            }

            default:
            {
                string str = windowfunction::colType2String[ti.dtype];
                throw IDBExcept(IDBErrorInfo::instance()->errorMsg(ERR_WF_INVALID_BOUND_TYPE, str),
                                ERR_WF_INVALID_BOUND_TYPE);
                break;
            }
        }
    }

    return fb;
}


boost::shared_ptr<FrameBound> WindowFunctionStep::parseFrameBoundRange(const execplan::WF_Boundary& b,
        const map<uint64_t, uint64_t>& m,
        const vector<SRCP>& o,
        JobInfo& jobInfo)
{
    boost::shared_ptr<FrameBound> fb;

    if (b.fFrame == WF_CURRENT_ROW)
    {
        fb.reset(new FrameBoundRange(WF__CURRENT_ROW));
        return fb;
    }

    bool isConstant = false;
    bool isNull = false;
    Row  dummy;
    ConstantColumn* cc = dynamic_cast<ConstantColumn*>(b.fVal.get());

    if (cc != NULL)
    {
        isConstant = true;
        double val = cc->getDoubleVal(dummy, isNull);

        if (val < 0 || isNull)
        {
            string str("NULL");

            if (!isNull)
            {
                ostringstream oss;
                oss << val;
                str = oss.str();
            }

            throw IDBExcept(IDBErrorInfo::instance()->errorMsg(ERR_WF_BOUND_OUT_OF_RANGE, str),
                            ERR_WF_BOUND_OUT_OF_RANGE);
        }
    }

    int type = 0;
    vector<uint64_t> ids;
    vector<int> index;
    ids.push_back(getTupleKey(jobInfo, o[0]));
    index.push_back(getColumnIndex(o[0], m, jobInfo));

    if (isConstant)
    {
        type = (b.fFrame == WF_PRECEDING) ? WF__CONSTANT_PRECEDING : WF__CONSTANT_FOLLOWING;
        ids.push_back(-1);   // dummy, n/a for constant
        index.push_back(-1); // dummy, n/a for constant
    }
    else
    {
        type = (b.fFrame == WF_PRECEDING) ? WF__EXPRESSION_PRECEDING : WF__EXPRESSION_FOLLOWING;
        ids.push_back(getTupleKey(jobInfo, b.fVal));
        index.push_back(getColumnIndex(b.fVal, m, jobInfo));
    }

    ids.push_back(getTupleKey(jobInfo, b.fBound));
    index.push_back(getColumnIndex(b.fBound, m, jobInfo));

    FrameBoundRange* fbr = NULL;
    TupleInfo ti = getTupleInfo(ids[0], jobInfo);
    bool asc = o[0]->asc();
    bool nlf = o[0]->nullsFirst();

    switch (ti.dtype)
    {
        case execplan::CalpontSystemCatalog::TINYINT:
        case execplan::CalpontSystemCatalog::SMALLINT:
        case execplan::CalpontSystemCatalog::MEDINT:
        case execplan::CalpontSystemCatalog::INT:
        case execplan::CalpontSystemCatalog::BIGINT:
        case execplan::CalpontSystemCatalog::DECIMAL:
        {
            if (isConstant)
            {
                int64_t v = cc->getIntVal(dummy, isNull);
                fbr = new FrameBoundConstantRange<int64_t>(type, asc, nlf, &v);
                fbr->isZero((v == 0));
            }
            else
            {
                fbr = new FrameBoundExpressionRange<int64_t>(type, asc, nlf);
            }

            break;
        }

        case execplan::CalpontSystemCatalog::DOUBLE:
        case execplan::CalpontSystemCatalog::UDOUBLE:
        {
            if (isConstant)
            {
                double v = cc->getDoubleVal(dummy, isNull);
                fbr = new FrameBoundConstantRange<double>(type, asc, nlf, &v);
                fbr->isZero((v == 0.0));
            }
            else
            {
                fbr = new FrameBoundExpressionRange<double>(type, asc, nlf);
            }

            break;
        }

        case execplan::CalpontSystemCatalog::FLOAT:
        case execplan::CalpontSystemCatalog::UFLOAT:
        {
            if (isConstant)
            {
                float v = cc->getFloatVal(dummy, isNull);
                fbr = new FrameBoundConstantRange<float>(type, asc, nlf, &v);
                fbr->isZero((v == 0.0));
            }
            else
            {
                fbr = new FrameBoundExpressionRange<float>(type, asc, nlf);
            }

            break;
        }

        case execplan::CalpontSystemCatalog::UTINYINT:
        case execplan::CalpontSystemCatalog::USMALLINT:
        case execplan::CalpontSystemCatalog::UMEDINT:
        case execplan::CalpontSystemCatalog::UINT:
        case execplan::CalpontSystemCatalog::UBIGINT:
        case execplan::CalpontSystemCatalog::UDECIMAL:
        case execplan::CalpontSystemCatalog::DATE:
        case execplan::CalpontSystemCatalog::DATETIME:
        case execplan::CalpontSystemCatalog::TIME:
        case execplan::CalpontSystemCatalog::TIMESTAMP:
        {
            if (isConstant)
            {
                uint64_t v = cc->getUintVal(dummy, isNull);
                fbr = new FrameBoundConstantRange<uint64_t>(type, asc, nlf, &v);
                fbr->isZero((v == 0));
            }
            else
            {
                fbr = new FrameBoundExpressionRange<uint64_t>(type, asc, nlf);
            }

            break;
        }

        default:
        {
            string str = windowfunction::colType2String[ti.dtype];
            throw IDBExcept(IDBErrorInfo::instance()->errorMsg(ERR_WF_INVALID_BOUND_TYPE, str),
                            ERR_WF_INVALID_BOUND_TYPE);
            break;
        }
    }

    fbr->setTupleId(ids);
    fbr->setIndex(index);
    fb.reset(fbr);

    return fb;
}


boost::shared_ptr<FrameBound> WindowFunctionStep::parseFrameBound(const execplan::WF_Boundary& b,
        const map<uint64_t, uint64_t>& m,
        const vector<SRCP>& o,
        const boost::shared_ptr<EqualCompData>& p,
        JobInfo& j,
        bool rows,
        bool s)
{
    boost::shared_ptr<FrameBound> fb;

    switch (b.fFrame)
    {
        case WF_UNBOUNDED_PRECEDING:
        {
            fb.reset(new FrameBound(WF__UNBOUNDED_PRECEDING));
            break;
        }

        case WF_UNBOUNDED_FOLLOWING:
        {
            fb.reset(new FrameBound(WF__UNBOUNDED_FOLLOWING));
            break;
        }

        case WF_CURRENT_ROW:
        case WF_PRECEDING:
        case WF_FOLLOWING:
        {
            if (rows)
            {
                fb = parseFrameBoundRows(b, m, j);
            }
            else
            {
                fb = parseFrameBoundRange(b, m, o, j);
            }

            break;
        }

        default:  //  unknown
        {
            throw IDBExcept(IDBErrorInfo::instance()->errorMsg(ERR_WF_UNKNOWN_BOUND, b.fFrame),
                            ERR_WF_UNKNOWN_BOUND);
            break;
        }
    }

    fb->peer(p);
    fb->start(s);

    return fb;
}


void WindowFunctionStep::updateWindowCols(ReturnedColumn* rc,
        const map<uint64_t, uint64_t>& m,
        JobInfo& jobInfo)
{
    if (rc == NULL)
        return;

    ArithmeticColumn*     ac = dynamic_cast<ArithmeticColumn*>(rc);
    FunctionColumn*       fc = dynamic_cast<FunctionColumn*>(rc);
    SimpleFilter*         sf = dynamic_cast<SimpleFilter*>(rc);
    WindowFunctionColumn* wc = dynamic_cast<WindowFunctionColumn*>(rc);

    if (wc)
    {
        uint64_t key = getExpTupleKey(jobInfo, wc->expressionId());
        map<uint64_t, uint64_t>::const_iterator j = m.find(key);

        if (j == m.end())
        {
            string name = jobInfo.keyInfo->tupleKeyToName[key];
            cerr << name << " is not in tuple, key=" << key << endl;
            throw IDBExcept(IDBErrorInfo::instance()->errorMsg(ERR_WF_COLUMN_MISSING, name),
                            ERR_WF_COLUMN_MISSING);
        }

        wc->inputIndex(j->second);
    }
    else if (ac)
    {
        updateWindowCols(ac->expression(), m, jobInfo);
    }
    else if (fc)
    {
        funcexp::FunctionParm parms = fc->functionParms();

        for (vector<execplan::SPTP>::iterator i = parms.begin(); i < parms.end(); i++)
            updateWindowCols(i->get(), m, jobInfo);
    }
    else if (sf)
    {
        updateWindowCols(sf->lhs(), m, jobInfo);
        updateWindowCols(sf->rhs(), m, jobInfo);
    }
}


void WindowFunctionStep::updateWindowCols(ParseTree* pt,
        const map<uint64_t, uint64_t>& m,
        JobInfo& jobInfo)
{
    if (pt == NULL)
        return;

    updateWindowCols(pt->left(), m, jobInfo);
    updateWindowCols(pt->right(), m, jobInfo);

    TreeNode* tn = pt->data();
    ArithmeticColumn*     ac = dynamic_cast<ArithmeticColumn*>(tn);
    FunctionColumn*       fc = dynamic_cast<FunctionColumn*>(tn);
    SimpleFilter*         sf = dynamic_cast<SimpleFilter*>(tn);
    WindowFunctionColumn* wc = dynamic_cast<WindowFunctionColumn*>(tn);

    if (wc)
    {
        uint64_t key = getExpTupleKey(jobInfo, wc->expressionId());
        map<uint64_t, uint64_t>::const_iterator j = m.find(key);

        if (j == m.end())
        {
            string name = jobInfo.keyInfo->tupleKeyToName[key];
            cerr << name << " is not in tuple, key=" << key << endl;
            throw IDBExcept(IDBErrorInfo::instance()->errorMsg(ERR_WF_COLUMN_MISSING, name),
                            ERR_WF_COLUMN_MISSING);
        }

        wc->inputIndex(j->second);
    }
    else if (ac)
    {
        updateWindowCols(ac->expression(), m, jobInfo);
    }
    else if (fc)
    {
        funcexp::FunctionParm parms = fc->functionParms();

        for (vector<execplan::SPTP>::iterator i = parms.begin(); i < parms.end(); i++)
            updateWindowCols(i->get(), m, jobInfo);
    }
    else if (sf)
    {
        updateWindowCols(sf->lhs(), m, jobInfo);
        updateWindowCols(sf->rhs(), m, jobInfo);
    }
}


void WindowFunctionStep::sort(std::vector<RowPosition>::iterator v, uint64_t n)
{
    // recursive function termination condition.
    if (n < 2 || cancelled())
        return;

    RowPosition                   p = *(v + n / 2); // pivot value
    vector<RowPosition>::iterator l = v;            // low   address
    vector<RowPosition>::iterator h = v + (n - 1);  // high  address

    while (l <= h && !cancelled())
    {
        // Can use while here, but need check boundary and cancel status.
        if (fQueryOrderBy->operator()(getPointer(*l), getPointer(p)))
        {
            l++;
        }
        else if (fQueryOrderBy->operator()(getPointer(p), getPointer(*h)))
        {
            h--;
        }
        else
        {
            RowPosition t = *l;    // temp value for swap
            *l++ = *h;
            *h-- = t;
        }
    }

    sort(v, std::distance(v, h) + 1);
    sort(l, std::distance(l, v) + n);
}


void WindowFunctionStep::printCalTrace()
{
    time_t t = time (0);
    char timeString[50];
    ctime_r (&t, timeString);
    timeString[strlen (timeString ) - 1] = '\0';
    ostringstream logStr;
    logStr  << "ses:" << fSessionId << " st: " << fStepId << " finished at " << timeString
            << "; total rows returned-" << fRowsReturned << endl
            << "\t1st read " << dlTimes.FirstReadTimeString()
            << "; EOI " << dlTimes.EndOfInputTimeString() << "; runtime-"
            << JSTimeStamp::tsdiffstr(dlTimes.EndOfInputTime(), dlTimes.FirstReadTime())
            << "s;\n\tUUID " << uuids::to_string(fStepUuid) << endl
            << "\tJob completion status " << status() << endl;
    logEnd(logStr.str().c_str());
    fExtendedInfo += logStr.str();
    formatMiniStats();
}


void WindowFunctionStep::formatMiniStats()
{
    ostringstream oss;
    oss << "WFS "
        << "UM "
        << "- "
        << "- "
        << "- "
        << "- "
        << "- "
        << "- "
        << JSTimeStamp::tsdiffstr(dlTimes.EndOfInputTime(), dlTimes.FirstReadTime()) << " "
        << fRowsReturned << " ";
    fMiniInfo += oss.str();
}


}   //namespace
// vim:ts=4 sw=4: