Big-Data-HPC-Platform/ecl/hqlcpp/hqlstep.cpp

/*##############################################################################

    HPCC SYSTEMS software Copyright (C) 2012 HPCC Systems®.

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

       http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
############################################################################## */
#include "jliball.hpp"
#include "hql.hpp"

#include "platform.h"
#include "jlib.hpp"
#include "jmisc.hpp"
#include "jstream.ipp"
#include "jdebug.hpp"

#include "hql.hpp"
#include "hqlthql.hpp"
#include "hqlhtcpp.ipp"
#include "hqlttcpp.ipp"
#include "hqlutil.hpp"
#include "hqlpmap.hpp"

#include "hqlwcpp.hpp"
#include "hqlcpputil.hpp"
#include "hqltcppc.ipp"
#include "hqlopt.hpp"
#include "hqlfold.hpp"
#include "hqlcerrors.hpp"
#include "hqlcatom.hpp"
#include "hqllib.ipp"
#include "hqlresource.hpp"
#include "hqlregex.ipp"
#include "hqlsource.ipp"
#include "hqlcse.ipp"
#include "hqlgraph.ipp"
#include "hqlscope.hpp"
#include "hqlccommon.hpp"
#include "hqlcppds.hpp"

#include "eclhelper.hpp"

#include "deffield.hpp"

//===========================================================================

class SteppingCondition
{
public:
    SteppingCondition(bool _onlyEquality, IHqlExpression * _left, IHqlExpression * _right, IHqlExpression * _rowsid);

    bool extractSteppingCondition(IHqlExpression * expr, IHqlExpression * sortOrder);
    IHqlExpression * createEqualitySortList();
    bool matchedAny() { return equalities.ordinality() != 0 || compareLhs; }
    IHqlExpression * queryRangeLeftSelector() { return compareLhs; }
    IHqlExpression * queryExtraFilter() { return extraCompare; }
    IHqlExpression * queryGlobalCompare() { return globalCompare; }

    IHqlExpression * getMaxLeftBeforeRight() { return getNegative(lhsDelta); }
    IHqlExpression * getMaxRightBeforeLeft() { return getNegative(rhsDelta); }

protected:
    void clearRangeMatch()
    {
        compareLhs.clear();
    }

    bool containsRowsLeft(IHqlExpression * expr);
    bool extractCondition(IHqlExpression * expr, IHqlExpression * searchField);
    bool extractCondition(HqlExprArray & args, IHqlExpression * searchField);
    bool extractComparison(IHqlExpression * lhs, IHqlExpression * rhs, IHqlExpression * searchField, bool isEqual = false);
    bool isLeftRightInvariant(IHqlExpression * expr);
    IHqlExpression * simplifyArgument(IHqlExpression * expr, SharedHqlExpr & delta, bool invert);

protected:
    HqlExprArray equalities;
    HqlExprAttr compareLhs;
    HqlExprAttr lhsDelta;               // left.x >= right.x + lhsDelta
    HqlExprAttr rhsDelta;               // right.x >= left.x + rhsDelta
    OwnedHqlExpr extraCompare;
    OwnedHqlExpr globalCompare;
    LinkedHqlExpr left;
    LinkedHqlExpr right;
    OwnedHqlExpr rowsLeftExpr;
    bool onlyEquality;
    bool explicitStepped;
};


SteppingCondition::SteppingCondition(bool _onlyEquality, IHqlExpression * _left, IHqlExpression * _right, IHqlExpression * _rowsid) : left(_left), right(_right), onlyEquality(_onlyEquality)
{
    explicitStepped = false;
    rowsLeftExpr.setown(createDataset(no_rows, LINK(left), LINK(_rowsid)));
}

IHqlExpression * SteppingCondition::createEqualitySortList()
{
    return createValueSafe(no_sortlist, makeSortListType(NULL), equalities);
}

bool SteppingCondition::extractSteppingCondition(IHqlExpression * expr, IHqlExpression * sortOrder)
{
    HqlExprArray args, stepArgs;
    expr->unwindList(args, no_and);

    explicitStepped = false;;
    ForEachItemIn(i1, args)
    {
        IHqlExpression & cur = args.item(i1);
        if (cur.getOperator() == no_assertstepped)
        {
            explicitStepped = true;
            cur.queryChild(0)->unwindList(stepArgs, no_and);
        }
    }

    //The merge order defines the order that the stepping fields are processed in.
    HqlExprArray order, expandedOrder;
    unwindChildren(order, sortOrder);
    expandRowSelectors(expandedOrder, order);
    bool foundStepped = explicitStepped;
    ForEachItemIn(i2, expandedOrder)
    {
        IHqlExpression * cur = &expandedOrder.item(i2);
        if (explicitStepped)
        {
            if (!extractCondition(stepArgs, cur))
            {
                StringBuffer s;
                if (cur->getOperator() == no_select)
                    s.append(cur->queryChild(1)->queryName());
                else
                    getExprECL(cur, s);
                throwError1(HQLERR_SteppingNotMatchSortCondition, s.str());
            }
            if (stepArgs.ordinality() == 0)
                break;
        }
        else
        {
            if (!extractCondition(args, cur))
                break;
            foundStepped = true;
        }
        if (compareLhs)
            break;
    }

    if (stepArgs.ordinality())
        throwError1(HQLERR_SteppingNotMatchSortCondition, "");

    //Walk the list of non stepped condition, and retain any that are dependent on rows(left)
    ForEachItemIn(i3, args)
    {
        IHqlExpression & cur = args.item(i3);
        if (cur.getOperator() != no_assertstepped)
        {
            if (containsRowsLeft(&cur))
                extendConditionOwn(globalCompare, no_and, LINK(&cur));
            else
                extendConditionOwn(extraCompare, no_and, LINK(&cur));
        }
    }

    return foundStepped;
}


bool SteppingCondition::containsRowsLeft(IHqlExpression * expr)
{
    OwnedHqlExpr null = createDataset(no_null, LINK(left->queryRecord()));
    OwnedHqlExpr replaceLeft = replaceExpression(expr, rowsLeftExpr, null);
    return replaceLeft != expr;
}

bool SteppingCondition::isLeftRightInvariant(IHqlExpression * expr)
{
    //MORE: is his good enough?
    OwnedHqlExpr replaceLeft = replaceSelector(expr, left, right);
    OwnedHqlExpr replaceRight = replaceSelector(expr, right, left);
    if (expr == replaceLeft && expr == replaceRight)
        return true;
    return false;
}


void adjustValue(SharedHqlExpr & total, IHqlExpression * value, bool invert)
{
    if (total)
        total.setown(adjustBoundIntegerValues(total, value, invert));
    else if (!invert)
        total.set(value);
    else
        total.setown(getNegative(value));
}


IHqlExpression * SteppingCondition::simplifyArgument(IHqlExpression * expr, SharedHqlExpr & delta, bool invert)
{
    for (;;)
    {
        switch (expr->getOperator())
        {
        case no_cast:
        case no_implicitcast:
            if (!castPreservesInformationAndOrder(expr))
                return expr;
            expr = expr->queryChild(0);
            break;
        case no_add:
            {
                IHqlExpression * lhs = expr->queryChild(0);
                IHqlExpression * rhs = expr->queryChild(1);
                if (isLeftRightInvariant(rhs))
                {
                    adjustValue(delta, rhs, invert);
                    expr = lhs;
                }
                else if (isLeftRightInvariant(lhs))
                {
                    adjustValue(delta, lhs, invert);
                    expr = rhs;
                }
                else
                    return expr;
                break;
            }
        case no_sub:
            {
                IHqlExpression * lhs = expr->queryChild(0);
                IHqlExpression * rhs = expr->queryChild(1);
                if (isLeftRightInvariant(rhs))
                {
                    adjustValue(delta, rhs, !invert);
                    expr = lhs;
                }
                else
                    return expr;
                break;
            }
        default:
            return expr;
        }
    }
}


bool SteppingCondition::extractComparison(IHqlExpression * lhs, IHqlExpression * rhs, IHqlExpression * searchField, bool isEqual)
{
    OwnedHqlExpr lhsSelect;
    OwnedHqlExpr delta;
    IHqlExpression * simpleLhs = simplifyArgument(lhs, delta, true);
    IHqlExpression * simpleRhs = simplifyArgument(rhs, delta, false);

    OwnedHqlExpr searchRightField = replaceSelector(searchField, left, right);
    if ((simpleLhs == searchField) && (simpleRhs == searchRightField))
    {
        compareLhs.set(searchField);
        if (!delta)
            delta.setown(getZero());
        if (lhsDelta)
        {
            StringBuffer s;
            throwError1(HQLERR_SteppedMultiRange, getExprECL(searchField,s).str());
        }
        lhsDelta.set(delta);
        if (isEqual)
            rhsDelta.setown(getNegative(delta));
        return true;
    }
    if ((simpleLhs == searchRightField) && (simpleRhs == searchField))
    {
        compareLhs.set(searchField);
        if (!delta)
            delta.setown(getZero());
        if (rhsDelta)
        {
            StringBuffer s;
            throwError1(HQLERR_SteppedMultiRange, getExprECL(searchRightField,s).str());
        }
        rhsDelta.set(delta);
        if (isEqual)
            lhsDelta.setown(getNegative(delta));
        return true;
    }
    return false;
}

bool SteppingCondition::extractCondition(IHqlExpression * expr, IHqlExpression * searchField)
{
    //Search for LEFT.someSelect = right.someSelect
    node_operator op = expr->getOperator();
    assertex(op != no_and);

    if (op == no_eq)
    {
        IHqlExpression * lhs = expr->queryChild(0);
        IHqlExpression * rhs = expr->queryChild(1);
        if (lhs == searchField)
        {
            OwnedHqlExpr replaced = replaceSelector(rhs, right, left);
            if (replaced == lhs)
            {
                equalities.append(*LINK(lhs));
                return true;
            }
        }
        if (rhs == searchField)
        {
            OwnedHqlExpr replaced = replaceSelector(lhs, right, left);
            if (replaced == rhs)
            {
                equalities.append(*LINK(rhs));
                return true;
            }
        }
    }

    if (!onlyEquality)
    {
        //left.x + d1 >= right.wpos + d2                (d1, d2 may be subtracted, and may be implicit casts in the expression)
        //normalize to left.x >= right.x + delta
        //left => maxRightAfterLeft = -delta;    right => maxRightBeforeLeft = -delta;

        switch (op)
        {
        case no_ge:
            return extractComparison(expr->queryChild(0), expr->queryChild(1), searchField);
        case no_le:
            return extractComparison(expr->queryChild(1), expr->queryChild(0), searchField);
        case no_between:
            if (extractComparison(expr->queryChild(0), expr->queryChild(1), searchField))
            {
                if (extractComparison(expr->queryChild(2), expr->queryChild(0), searchField))
                    return true;
                clearRangeMatch();
            }
            break;
        case no_eq:
            return extractComparison(expr->queryChild(0), expr->queryChild(1), searchField, true);
        }
    }
    return false;
}


bool SteppingCondition::extractCondition(HqlExprArray & args, IHqlExpression * searchField)
{
    assertex(!compareLhs);
    UnsignedArray matched;

    ForEachItemIn(i, args)
    {
        IHqlExpression & cur = args.item(i);
        if (extractCondition(&cur, searchField))
            matched.append(i);
    }

    if (compareLhs)
    {
        //Only matched in one direction
        if (!lhsDelta || !rhsDelta)
        {
            if (explicitStepped)
                throwError(HQLERR_SteppedRangeOnlyOneDirection);
            matched.kill();
        }
        else
        {
            ForEachItemInRev(i2, matched)
            {
                IHqlExpression & cur = args.item(matched.item(i2));
                extendConditionOwn(extraCompare, no_and, LINK(&cur));
            }
        }
    }

    ForEachItemInRev(i2, matched)
        args.remove(matched.item(i2));

    return matched.ordinality() != 0;
}

//---------------------------------------------------------------------------

void SteppingFieldSelection::expandTransform(IHqlExpression * expr)
{
    IHqlExpression * parent = expr->queryChild(0)->queryNormalizedSelector();

    TableProjectMapper mapper(expr);
    if (!mapper.isMappingKnown())
        throwError(HQLERR_CantProjectStepping);

    fields.setown(mapper.expandFields(fields, ds, parent));
    fields.setown(expandCreateRowSelectors(fields));
    ds.set(parent);
}

void SteppingFieldSelection::extractFields(SteppingFieldSelection & steppingFields)
{
    steppingFields.ds.set(ds);
    HqlExprArray args;
    ForEachChild(i, fields)
    {
        IHqlExpression * cur = fields->queryChild(i);
        args.append(*extractSelect(cur));
    }
    steppingFields.fields.setown(fields->clone(args));
}


static void throwTooComplexToStep(IHqlExpression * expr)
{
    StringBuffer ecl;
    getExprECL(expr, ecl, true, false);
    throwError1(HQLERR_TooComplexToStep, ecl.str());
}

IHqlExpression * SteppingFieldSelection::extractSelect(IHqlExpression * expr)
{
    for (;;)
    {
        switch (expr->getOperator())
        {
        case no_filepos:
        case no_file_logicalname:
            //MORE: We should really catch more problems like this...
            throwError(HQLERR_NoSteppingOnPayload);
        case no_cast:
        case no_implicitcast:
            {
                if (!castPreservesValueAndOrder(expr))
                {
                    switch (expr->queryChild(0)->getOperator())
                    {
                    case no_filepos:
                    case no_file_logicalname:
                        throwError(HQLERR_NoSteppingOnPayload);
                    default:
                        throwTooComplexToStep(expr);
                    }
                }
                expr = expr->queryChild(0);
                break;
            }
        case no_add:
        case no_sub:
            {
                //cope with biasing on indexes.
                IHqlExpression * rhs = expr->queryChild(1);
                switch (rhs->getOperator())
                {
                case no_constant:
                    break;
                default:
                    throwTooComplexToStep(expr);
                }
                expr = expr->queryChild(0);
                break;
            }
        case no_select:
            return LINK(expr);
        default:
            throwTooComplexToStep(expr);
        }
    }
}

void SteppingFieldSelection::gatherFieldOffsetSizes(HqlCppTranslator & translator, UnsignedArray & result)
{
    //A pseudo context in somewhere that will never be generated.
    BuildCtx ctx(*translator.queryCode(), _internal_Atom);
    ctx.addGroup();
    translator.bindTableCursor(ctx, ds, "x");

    CHqlBoundExpr bound;
    StringBuffer s;
    ForEachChild(i, fields)
    {
        IHqlExpression * cur = fields->queryChild(i);
        assertex(cur->getOperator() == no_select);
        Owned<IReferenceSelector> selector = translator.buildActiveReference(ctx, cur);
        selector->getOffset(ctx, bound);
        IValue * offsetValue = bound.expr->queryValue();
        if (offsetValue)
            result.append((unsigned)offsetValue->getIntValue());
        else
            throwError1(HQLERR_SteppedVariableOffset, getExprECL(cur, s).str());

        selector->getSize(ctx, bound);
        IValue * sizeValue = bound.expr->queryValue();
        if (sizeValue)
            result.append((unsigned)sizeValue->getIntValue());
        else
            throwError1(HQLERR_SteppedVariableSize, getExprECL(cur, s).str());
    }
}

IHqlExpression * SteppingFieldSelection::generateSteppingMeta(HqlCppTranslator & translator)
{
    OwnedHqlExpr normalFields = replaceSelector(fields, ds, queryActiveTableSelector());
    OwnedHqlExpr key = createAttribute(_steppedMeta_Atom, LINK(ds->queryRecord()->queryBody()), LINK(normalFields));
    BuildCtx declarectx(*translator.queryCode(), declareAtom);

    HqlExprAssociation * match = declarectx.queryMatchExpr(key);
    if (match)
        return match->queryExpr();

    BuildCtx classctx(declarectx);
    StringBuffer s, s2;
    StringBuffer memberName, offsetName;
    unique_id_t id = translator.getUniqueId();
    appendUniqueId(memberName.append("st"), id);
    appendUniqueId(offsetName.append("so"), id);

    UnsignedArray offsets;
    gatherFieldOffsetSizes(translator, offsets);

    unsigned lenOffsets = offsets.ordinality();
    s.clear();
    s.append("CFieldOffsetSize ").append(offsetName).append("[").append(lenOffsets/2).append("] = {");
    for (unsigned i=0; i < lenOffsets; i += 2)
    {
        if (i) s.append(",");
        s.append("{").append(offsets.item(i)).append(",").append(offsets.item(i+1)).append("}");
    }
    s.append("};");
    declarectx.setNextPriority(SteppedPrio);
    declarectx.addQuoted(s);

    //MORE: This might be better commoned up globally, depending of number of instances
    classctx.setNextPriority(SteppedPrio);
    classctx.addQuotedCompound(s.clear().append("struct C").append(memberName).append(" : public ISteppingMeta"), s2.append(" ").append(memberName).append(";").str());
    translator.doBuildUnsignedFunction(classctx, "getNumFields", lenOffsets/2);

    classctx.addQuoted(s.clear().append("virtual const CFieldOffsetSize * queryFields() override { return ").append(offsetName).append("; }"));

    //compare function.
    {
        StringBuffer compareName;
        translator.getUniqueId(compareName.append("c"));
        OwnedITypeInfo intType = makeIntType(4, true);
        OwnedHqlExpr result = createVariable("ret", LINK(intType));

        BuildCtx comparectx(classctx);
        comparectx.addQuotedCompoundLiteral("class Compare : public IRangeCompare", s2.clear().append(" ").append(compareName).append(";"));
        translator.doBuildUnsignedFunction(comparectx, "maxFields", lenOffsets/2);

        {
            MemberFunction func(translator, comparectx, "virtual int docompare(const void * _left,const void * _right, unsigned numFields) const override");
            func.ctx.addQuotedLiteral("const byte * left = (const byte *)_left;");
            func.ctx.addQuotedLiteral("const byte * right = (const byte *)_right;");
            func.ctx.addQuotedLiteral("int ret;");

            func.ctx.addQuoted(s.clear().append("if (numFields < 1) return 0;"));
            OwnedHqlExpr selSeq = createDummySelectorSequence();
            BoundRow * left = translator.bindTableCursor(func.ctx, ds, "left", no_left, selSeq);
            BoundRow * right = translator.bindTableCursor(func.ctx, ds, "right", no_right, selSeq);
            ForEachChild(i, fields)
            {
                IHqlExpression * cur = fields->queryChild(i);
                if (i)
                    func.ctx.addQuoted(s.clear().append("if (ret || (numFields < ").append(i+1).append(")) return ret;"));
                OwnedHqlExpr lhs = replaceSelector(cur, ds, left->querySelector());
                OwnedHqlExpr rhs = replaceSelector(cur, ds, right->querySelector());
                OwnedHqlExpr order = createValue(no_order, makeIntType(4, true), LINK(lhs), LINK(rhs));
                translator.buildAssignToTemp(func.ctx, result, order);
            }
            func.ctx.addReturn(result);
        }
    
        classctx.addQuoted(s.clear().append("virtual IRangeCompare * queryCompare() override { return &").append(compareName).append("; }"));
    }

    //distance function - very similar to compare
    {
        StringBuffer distanceName;
        translator.getUniqueId(distanceName.append("c"));
        OwnedITypeInfo intType = makeIntType(4, true);
        OwnedHqlExpr result = createVariable("ret", LINK(intType));

        BuildCtx distancectx(classctx);
        distancectx.addQuotedCompoundLiteral("class Distance : public IDistanceCalculator", s2.clear().append(" ").append(distanceName).append(";"));

        {
            MemberFunction func(translator, distancectx, "virtual unsigned getDistance(unsigned __int64 & distance, const void * _before, const void * _after, unsigned numFields) const override");
            func.ctx.addQuotedLiteral("const byte * before = (const byte *)_before;");
            func.ctx.addQuotedLiteral("const byte * after = (const byte *)_after;");

            OwnedHqlExpr selSeq = createDummySelectorSequence();
            OwnedITypeInfo distanceType = makeIntType(8, false);
            OwnedHqlExpr distanceExpr = createVariable("distance", LINK(distanceType));
            BoundRow * left = translator.bindTableCursor(func.ctx, ds, "before", no_left, selSeq);
            BoundRow * right = translator.bindTableCursor(func.ctx, ds, "after", no_right, selSeq);
            ForEachChild(i, fields)
            {
                IHqlExpression * cur = fields->queryChild(i);
                func.ctx.addQuoted(s.clear().append("if (numFields < ").append(i+1).append(") return DISTANCE_EXACT_MATCH;"));
                OwnedHqlExpr lhs = replaceSelector(cur, ds, left->querySelector());
                OwnedHqlExpr rhs = replaceSelector(cur, ds, right->querySelector());
                OwnedHqlExpr compare = createBoolExpr(no_ne, LINK(lhs), LINK(rhs));
                BuildCtx subctx(func.ctx);
                translator.buildFilter(subctx, compare);

                OwnedHqlExpr value;
                if (lhs->queryType()->isInteger())
                    value.setown(createValue(no_sub, LINK(distanceType), ensureExprType(rhs, distanceType), ensureExprType(lhs, distanceType)));
                else
                    value.setown(getSizetConstant(1));
                translator.buildAssignToTemp(subctx, distanceExpr, value);
                subctx.addQuotedF("return %u;", i+1);
            }

            func.ctx.addQuotedLiteral("return DISTANCE_EXACT_MATCH;");
        }
    
        classctx.addQuoted(s.clear().append("virtual IDistanceCalculator * queryDistance() override { return &").append(distanceName).append("; }"));
    }

    StringBuffer resultText;
    if (translator.queryOptions().spanMultipleCpp)
    {
        translator.createAccessFunctions(resultText, declarectx, SteppedPrio, "ISteppingMeta", memberName);
        resultText.append("()");
    }
    else
        resultText.append(memberName);

    OwnedHqlExpr func = createVariable(resultText.str(), makeVoidType());
    declarectx.associateExpr(key, func);
    return func;
}

void SteppingFieldSelection::generateSteppingMetaMember(HqlCppTranslator & translator, BuildCtx & ctx, const char * name)
{
    IHqlExpression * func = generateSteppingMeta(translator);

    StringBuffer s;
    s.clear().append("virtual ISteppingMeta * query").append(name).append("() override { return & ");
    translator.generateExprCpp(s, func);
    s.append(";}");
    ctx.addQuoted(s);
}

IHqlExpression * SteppingFieldSelection::invertTransform(IHqlExpression * expr, IHqlExpression * select)
{
    LinkedHqlExpr result = select;
    for (;;)
    {
        node_operator op = expr->getOperator();
        switch (op)
        {
        case no_cast:
        case no_implicitcast:
            {
                IHqlExpression * uncast = expr->queryChild(0);
                result.setown(ensureExprType(result, uncast->queryType()));
                expr = uncast;
                break;
            }
        case no_add:
        case no_sub:
            {
                node_operator newOp = (op == no_add) ? no_sub : no_add;
                IHqlExpression * rhs = expr->queryChild(1);
                result.setown(createValue(newOp, expr->getType(), LINK(result), LINK(rhs)));
                expr = expr->queryChild(0);
                break;
            }
        case no_select:
            return result.getLink();
        default:
            throwUnexpectedOp(op);
        }
    }
}


void SteppingFieldSelection::set(IHqlExpression * _ds, IHqlExpression * _fields)
{
    ds.set(_ds);
    fields.set(_fields);
}

void SteppingFieldSelection::setStepping(IHqlExpression * expr)
{
    ds.set(expr->queryNormalizedSelector());
    fields.set(expr->queryChild(1));
}


//---------------------------------------------------------------------------

bool HqlCppTranslator::buildNWayInputs(CIArrayOf<ABoundActivity> & inputs, BuildCtx & ctx, IHqlExpression * input)
{
    if (input->getOperator() == no_datasetlist)
    {
        IHqlExpression * record = input->queryChild(0);
        ForEachChild(i, input)
        {
            IHqlExpression * cur = input->queryChild(i);
            if (!recordTypesMatch(cur->queryRecord(), record))
                throwError(HQLERR_InconsistentNaryInput);
            inputs.append(*buildCachedActivity(ctx, cur));
        }
        return false;
    }

    inputs.append(*buildCachedActivity(ctx, input));
    return true;
}


ABoundActivity * HqlCppTranslator::doBuildActivityRowsetRange(BuildCtx & ctx, IHqlExpression * expr, IHqlExpression * rowset, IHqlExpression * inputSelection)
{
    bool isNWayInput;
    ThorActivityKind kind;
    const char * argName;
    CIArrayOf<ABoundActivity> inputs;
    IHqlExpression * graphId = NULL;

    switch (rowset->getOperator())
    {
    case no_getgraphloopresultset:
        {
            kind = TAKnwaygraphloopresultread;
            argName = "NWayGraphLoopResultRead";
            isNWayInput = true;
            graphId = rowset->queryChild(1);
            break;
        }
    case no_datasetlist:
        {
            kind = TAKnwayinput;
            argName = "NWayInput";
            isNWayInput = false;
            ForEachChild(i, rowset)
                inputs.append(*buildCachedActivity(ctx, rowset->queryChild(i)));
            break;
        }
    default:
        throwError(HQLERR_UnsupportedRowsetRangeParam);
    }

    Owned<ActivityInstance> instance = new ActivityInstance(*this, ctx, kind, expr, argName);
    buildActivityFramework(instance);

    buildInstancePrefix(instance);

    OwnedITypeInfo castType = makeSetType(LINK(unsignedType));
    OwnedHqlExpr castList = ensureExprType(inputSelection, castType);
    OwnedHqlExpr normalized = normalizeListCasts(castList);

    {
        MemberFunction func(*this, instance->startctx, "virtual void getInputSelection(bool & __isAllResult, size32_t & __lenResult, void * & __result) override");
        doBuildFunctionReturn(func.ctx, castType, normalized);
    }

    if ((kind == TAKnwaygraphloopresultread) && isGrouped(rowset))
        doBuildBoolFunction(instance->classctx, "grouped", true);
    if (graphId && targetRoxie())
        addGraphIdAttribute(instance, ctx, graphId);

    buildInstanceSuffix(instance);

    ForEachItemIn(idx2, inputs)
        buildConnectInputOutput(ctx, instance, &inputs.item(idx2), 0, idx2, NULL, isNWayInput);

    return instance->getBoundActivity();
}

ABoundActivity * HqlCppTranslator::doBuildActivityRowsetRange(BuildCtx & ctx, IHqlExpression * expr)
{
    IHqlExpression * ds = expr->queryChild(0);
    IHqlExpression * inputSelection = expr->queryChild(1);
    return doBuildActivityRowsetRange(ctx, expr, ds, inputSelection);
}


ABoundActivity * HqlCppTranslator::doBuildActivityRowsetIndex(BuildCtx & ctx, IHqlExpression * expr)
{
    IHqlExpression * dataset = expr->queryChild(0);
    if (dataset->getOperator() == no_getgraphloopresultset)
    {
        throwUnexpected();      // this should have been translated elsewhere...
        OwnedHqlExpr newExpr = createDataset(no_getgraphloopresult, LINK(dataset->queryRecord()), createComma(LINK(dataset->queryChild(1)), LINK(expr->queryChild(1))));
        return buildActivity(ctx, newExpr, false);
    }

    CIArrayOf<ABoundActivity> inputs;
    bool isNWayInput = buildNWayInputs(inputs, ctx, dataset);

    Owned<ActivityInstance> instance = new ActivityInstance(*this, ctx, TAKnwayselect, expr, "NWaySelect");
    buildActivityFramework(instance);

    buildInstancePrefix(instance);

    doBuildUnsignedFunction(instance->startctx, "getInputIndex", expr->queryChild(1));

    buildInstanceSuffix(instance);

    ForEachItemIn(idx2, inputs)
        buildConnectInputOutput(ctx, instance, &inputs.item(idx2), 0, idx2, NULL, isNWayInput);

    return instance->getBoundActivity();
}



ABoundActivity * HqlCppTranslator::doBuildActivityNWayMerge(BuildCtx & ctx, IHqlExpression * expr)
{
    IHqlExpression * dataset = expr->queryChild(0);
    CIArrayOf<ABoundActivity> inputs;
    bool isNWayInput = buildNWayInputs(inputs, ctx, dataset);

    Owned<ActivityInstance> instance = new ActivityInstance(*this, ctx, TAKnwaymerge, expr, "NWayMerge");
    buildActivityFramework(instance);

    buildInstancePrefix(instance);

    IHqlExpression * sortOrder = expr->queryChild(1);

    //NOTE: left is used instead of dataset in sort list
    DatasetReference dsRef(dataset, no_left, querySelSeq(expr));
    buildCompareFuncHelper(*this, *instance, "compare", sortOrder, dsRef);

    if (expr->hasAttribute(dedupAtom))
        doBuildBoolFunction(instance->classctx, "dedup", true);

    SteppingFieldSelection stepping;
    IHqlExpression * left = dsRef.querySelector();
    stepping.set(left, sortOrder);
    stepping.generateSteppingMetaMember(*this, instance->classctx, "SteppingMeta");

    buildInstanceSuffix(instance);

    ForEachItemIn(idx2, inputs)
        buildConnectInputOutput(ctx, instance, &inputs.item(idx2), 0, idx2, NULL, isNWayInput);

    return instance->getBoundActivity();
}



ABoundActivity * HqlCppTranslator::doBuildActivityNWayMergeJoin(BuildCtx & ctx, IHqlExpression * expr)
{
    node_operator op = expr->getOperator();
    if (targetThor() && !isLocalActivity(expr) && !isGroupedActivity(expr) && !insideChildQuery(ctx))
    {
        //Should default to an error in a later version, but LOCAL wasn't allowed on MERGEJOIN so make a warning for now.
        reportWarning(CategoryUnexpected, SeverityUnknown, NULL, ECODETEXT(HQLWRN_OnlyLocalMergeJoin), getOpString(op));
        OwnedHqlExpr localExpr = appendLocalAttribute(expr);
        return doBuildActivityNWayMergeJoin(ctx, localExpr);
    }

    IHqlExpression * dataset = expr->queryChild(0);
    CIArrayOf<ABoundActivity> inputs;
    bool isNWayInput = buildNWayInputs(inputs, ctx, dataset);

    ThorActivityKind kind = (op == no_mergejoin) ? TAKnwaymergejoin : TAKnwayjoin;
    Owned<ActivityInstance> instance = new ActivityInstance(*this, ctx, kind, expr, "NWayMergeJoin");
    buildActivityFramework(instance);

    buildInstancePrefix(instance);

    IHqlExpression * mofn = expr->queryAttribute(mofnAtom);
    bool leftonly = expr->hasAttribute(leftonlyAtom);
    bool leftouter = expr->hasAttribute(leftouterAtom);
    IHqlExpression * selSeq = querySelSeq(expr);
    IHqlExpression * rowsid = expr->queryAttribute(_rowsid_Atom);
    IHqlExpression * transform = (op == no_nwayjoin) ? expr->queryChild(2) : NULL;
    IHqlExpression * sortOrder = (op == no_nwayjoin) ? expr->queryChild(3) : expr->queryChild(2);

    OwnedHqlExpr left = createSelector(no_left, dataset, selSeq);
    OwnedHqlExpr right = createSelector(no_right, dataset, selSeq);

    SteppingCondition stepCondition(false, left, right, rowsid);
    stepCondition.extractSteppingCondition(expr->queryChild(1), sortOrder);

    if (!stepCondition.matchedAny())
        throwError(HQLERR_JoinNotMatchSortCondition);

    OwnedHqlExpr equalityList = stepCondition.createEqualitySortList();
    IHqlExpression * rangeSelect = stepCondition.queryRangeLeftSelector();
    IHqlExpression * internalFlags = queryAttributeChild(expr, internalFlagsAtom, 0);
    IHqlExpression * skew = expr->queryAttribute(skewAtom);

    //Now generate all the helper functions....
    bool createClearRow = true;//(!leftouter && !leftonly);
    StringBuffer flags;
    flags.append("|MJFhasdistance");
    if (leftouter)
        flags.append("|MJFleftouter");
    else if (leftonly)
        flags.append("|MJFleftonly");
    else if (mofn)
        flags.append("|MJFmofn");
    else
        flags.append("|MJFinner");

    if (expr->hasAttribute(dedupAtom)) flags.append("|MJFdedup");
    if (expr->hasAttribute(steppedAtom)) flags.append("|MJFstepped");
    if (transform) flags.append("|MJFtransform");
    if (rangeSelect) flags.append("|MJFhasrange");
    if (expr->hasAttribute(assertAtom) && generateAsserts()) flags.append("|MJFassertsorted");
    if (stepCondition.queryGlobalCompare()) flags.append("|MJFglobalcompare");
    if (createClearRow) flags.append("|MJFhasclearlow");
    if (skew) flags.append("|MJFhaspartition");
    if (internalFlags) flags.append("|").append(getIntValue(internalFlags, 0));

    if (flags.length())
        doBuildUnsignedFunction(instance->classctx, "getJoinFlags", flags.str()+1);


    //NOTE: left is used instead of dataset in sort list
    DatasetReference leftRef(dataset, no_left, querySelSeq(expr));

    unsigned numEqualFields = equalityList->numChildren();
    doBuildUnsignedFunction(instance->classctx, "numEqualFields", numEqualFields);
    doBuildUnsignedFunction(instance->classctx, "numOrderFields", sortOrder->numChildren());

    //virtual ICompare * queryEqualCompare()
    buildCompareFuncHelper(*this, *instance, "equalCompare", equalityList, leftRef);

    //virtual ICompareEq * queryExactCompare()
    {
        buildCompareEqMember(instance->classctx, "EqualCompareEq", equalityList, leftRef);
    }

    //virtual ICompareEq * queryPartitionCompareEq()
    if (skew)
    {
        HqlExprArray skewArgs;
        unwindChildren(skewArgs, skew);
        OwnedHqlExpr skewOrder = createSortList(skewArgs);
        DatasetReference datasetRef(dataset);
        buildCompareEqMember(instance->classctx, "PartitionCompareEq", skewOrder, leftRef);
    }

    //virtual ISteppingMeta * querySteppingMeta()
    {
        SteppingFieldSelection stepping;
        stepping.set(left, sortOrder);
        stepping.generateSteppingMetaMember(*this, instance->classctx, "SteppingMeta");
    }

    //virtual IOutputMetaData * queryInputMeta()
    {
        MetaInstance inputmeta(*this, dataset->queryRecord(), isGrouped(dataset));
        buildMetaInfo(inputmeta);

        StringBuffer s;
        s.append("virtual IOutputMetaData * queryInputMeta() override { return &").append(inputmeta.queryInstanceObject()).append("; }");
        instance->classctx.addQuoted(s);
    }

    //NOTE: left is used instead of dataset in sort list
    //virtual ICompare * queryMergeCompare()
    buildCompareFuncHelper(*this, *instance, "mergeCompare", sortOrder, leftRef);

    if (createClearRow)
    {
        BuildCtx funcctx(instance->startctx);
        OwnedHqlExpr func = getClearRecordFunction(dataset->queryRecord(), -1);
        StringBuffer s;
        generateExprCpp(s.append("virtual size32_t createLowInputRow(ARowBuilder & crSelf) override { return "), func).append("(crSelf, ctx); }");
        funcctx.addQuoted(s);
    }

    if (rangeSelect)
    {
        OwnedITypeInfo rangeType = makeIntType(8, false);
        OwnedITypeInfo distanceType = makeIntType(8, true);
        OwnedHqlExpr rangeValue = ensureExprType(rangeSelect, rangeType);
        OwnedHqlExpr bias;
        if (rangeSelect->queryType()->isSigned())
        {
            bias.setown(getHozedBias(rangeSelect->queryType()));
            rangeValue.setown(createValue(no_add, rangeValue->getType(), LINK(rangeValue), ensureExprType(bias, rangeType)));
        }

        if (sortOrder->numChildren() != numEqualFields + 1)
            throwError(HQLERR_SortOrderMustMatchJoinFields);

        //virtual unsigned __int64 extractRangeValue(const void * input);               // distance is assumed to be unsigned, code generator must bias if not true.
        {
            MemberFunction func(*this, instance->startctx, "virtual unsigned __int64 extractRangeValue(const void * _left) override");
            func.ctx.addQuotedLiteral("const byte * left = (const byte *)_left;");
            bindTableCursor(func.ctx, dataset, "left", no_left, selSeq);
            buildReturn(func.ctx, rangeValue);
        }

        //virtual void adjustRangeValue(void * self, const void * input, __int64 delta);        // implementation must ensure field doesn't go -ve.
        {
            MemberFunction func(*this, instance->startctx, "virtual void adjustRangeValue(ARowBuilder & crSelf, const void * _left, __int64 delta) override");
            ensureRowAllocated(func.ctx, "crSelf");
            func.ctx.addQuotedLiteral("const byte * left = (const byte *)_left;");

            BoundRow * self = bindSelf(func.ctx, dataset, "crSelf");
            bindTableCursor(func.ctx, dataset, "left", no_left, selSeq);
            ForEachChild(i, equalityList)
            {
                IHqlExpression * cur = equalityList->queryChild(i);
                OwnedHqlExpr target = replaceSelector(cur, left, self->querySelector());
                buildAssign(func.ctx, target, cur);
            }
            OwnedHqlExpr target = replaceSelector(rangeSelect, left, self->querySelector());
            OwnedHqlExpr delta = createVariable("delta", LINK(distanceType));
            OwnedHqlExpr castDelta = ensureExprType(delta, rangeType);
            OwnedHqlExpr minusDelta = getNegative(delta);
            OwnedHqlExpr cond = createBoolExpr(no_or, 
                                            createBoolExpr(no_ge, LINK(delta), ensureExprType(queryZero(), distanceType)),
                                            createBoolExpr(no_ge, LINK(rangeValue), ensureExprType(minusDelta, rangeType)));
            OwnedHqlExpr firstValue = bias ? getNegative(bias) : getZero();
            OwnedHqlExpr assignValue = createValue(no_if, rangeSelect->getType(), 
                                                          LINK(cond), 
                                                          createValue(no_add, rangeSelect->getType(), LINK(rangeSelect), ensureExprType(delta, rangeSelect->queryType())),
                                                          ensureExprType(firstValue, rangeSelect->queryType()));
            buildAssign(func.ctx, target, assignValue);
        }

        //virtual __int64 maxRightBeforeLeft()
        {
            MemberFunction func(*this, instance->startctx, "virtual __int64 maxRightBeforeLeft() override");
            OwnedHqlExpr mrbl = stepCondition.getMaxRightBeforeLeft();
            buildReturn(func.ctx, mrbl);
        }

        //virtual __int64 maxLeftBeforeRight()
        {
            MemberFunction func(*this, instance->startctx, "virtual __int64 maxLeftBeforeRight() override");
            OwnedHqlExpr mlbr = stepCondition.getMaxLeftBeforeRight();
            buildReturn(func.ctx, mlbr);
        }
    }

    //virtual ICompareEq * queryNonSteppedCompare()
    IHqlExpression * compare = stepCondition.queryExtraFilter();
    if (compare)
        buildCompareEqMemberLR(instance->nestedctx, "NonSteppedCompare", compare, dataset, dataset, selSeq);

    //virtual INaryCompareEq * queryGlobalCompare() = 0;
    IHqlExpression * globalCompare = stepCondition.queryGlobalCompare();
    if (globalCompare)
        buildNaryCompareMember(instance->startctx, "GlobalCompare", globalCompare, dataset, selSeq, rowsid);

    //virtual size32_t transform(ARowBuilder & crSelf, unsigned _num, const void * * _rows)
    if (transform)
    {
        MemberFunction func(*this, instance->startctx, "virtual size32_t transform(ARowBuilder & crSelf, unsigned numRows, const void * * _rows) override");
        ensureRowAllocated(func.ctx, "crSelf");
        func.ctx.addQuotedLiteral("const unsigned char * left = (const unsigned char *) _rows[0];");
        func.ctx.addQuotedLiteral("const unsigned char * right = (const unsigned char *) _rows[1];");
        func.ctx.addQuotedLiteral("const byte * * rows = (const byte * *) _rows;");

        bindTableCursor(func.ctx, dataset, "left", no_left, selSeq);
        bindTableCursor(func.ctx, dataset, "right", no_right, selSeq);
        bindRows(func.ctx, no_left, selSeq, rowsid, dataset, "numRows", "rows", options.mainRowsAreLinkCounted);
        BoundRow * selfCursor = bindSelf(func.ctx, expr, "crSelf");

        associateSkipReturnMarker(func.ctx, queryZero(), selfCursor);
        doTransform(func.ctx, transform, selfCursor);
        buildReturnRecordSize(func.ctx, selfCursor);
    }

    if (mofn)
    {
        doBuildUnsignedFunction(instance->startctx, "getMinMatches", mofn->queryChild(0));
        if (queryRealChild(mofn, 1))
            doBuildUnsignedFunction(instance->startctx, "getMaxMatches", mofn->queryChild(1));
    }

    if (leftonly)
    {
        //Create a function to apply a delta to the last field, it assumes that overflow isn't going to be a problem.
        IHqlExpression * lastJoinField = equalityList->queryChild(numEqualFields-1);
        if (lastJoinField->queryType()->isInteger())
        {
            MemberFunction func(*this, instance->startctx, "virtual bool createNextJoinValue(ARowBuilder & crSelf, const void * _value) override");
            ensureRowAllocated(func.ctx, "crSelf");
            func.ctx.addQuotedLiteral("const byte * value = (const byte *)_value;");

            BoundRow * self = bindSelf(func.ctx, dataset, "crSelf");
            bindTableCursor(func.ctx, dataset, "value", no_left, selSeq);
            ForEachChild(i, equalityList)
            {
                IHqlExpression * cur = equalityList->queryChild(i);
                OwnedHqlExpr target = replaceSelector(cur, left, self->querySelector());
                LinkedHqlExpr source = cur;
                if (i == numEqualFields-1)
                    source.setown(adjustValue(cur, 1));
                buildAssign(func.ctx, target, source);
            }
            buildReturn(func.ctx, queryBoolExpr(true));
        }
    }


    buildInstanceSuffix(instance);

    ForEachItemIn(idx2, inputs)
        buildConnectInputOutput(ctx, instance, &inputs.item(idx2), 0, idx2, NULL, isNWayInput);

    return instance->getBoundActivity();
}



//---------------------------------------------------------------------------


/*

  Stepping info.
  Assume we have 
  a) an index read, stepped on [doc, wpos, wip]
  b) an index read, stepped on [doc, wpos]
  c) an index read, stepped on [doc, wpos, wip]
  d) mergejoin(a,b, merge[doc, wpos, wip], left.doc = right.doc));
  e) join(d, c, stepped(left.doc = right.doc, right.wpos in range left.wpos - 5, left.wpos + 10), sorted([doc, wpos, wip]);
  f) SORT(e, [doc, wpos, wip], RANGE(left.wpos - right.wpos between [-5, 5]))       
     // could push top and right scope for range, but not very nice..., introduce a new no_sort keyword regardless of syntax.


We have
    a) static stepping = [doc,wpos,wip], dynamic matchee
    b) static stepping = [doc, wpos], dyamic matches
    c) same as a
    d) static stepping = [doc, wpos, wip]
       dynamic = dynamic for input#0 intersected with own static stepping.
       because a merge, all fields used in the merge can be stepped.
    e) static = [doc, wpos], because those are the conditions used in the join condition, and each of those values is either assigned left.x or right.x inside the transform
       sorting = [doc], or possibly [doc, wpos] if assignment self.wpos = left.wpos in transform
       stepping on [doc, wpos] is handled by adjusting the requested value by the maximum (delta1, delta2), since it is either assigned left/right.  This should be a separate constant
       so the self.x := left.x can be optimized to delta1, but fairly insignificant.
    f) static = [doc, wpos] - from sort criteria, and field referenced in the proximity condition
       dynamic = [doc, wpos] after intersection with output from e.
       sorted by [doc, wpos, wip] again.

More on JOIN:
  * Write code to allow nesting ((a JOIN b) JOIN c) with different deltas for each level.
  * Do all the seeks before creating any of the records.  Probably need to find the first candidate in parallel, and then recursively create the transforms.

    seek(n) = seek(applyDelta(min(values[1..n-1], minRightBeforeLeft);
    if fail, adjust match, by minRightBeforeLeft, and start seeking on 1 again.
    once you've got a match, go off and create the instances.

  For arbitrary nesting
    (a w/x b) w/y (c w/z d)
    Seek(a)
    seek(b, matcha-x);
    seek(c, min(a,b)-(y+z));
    seek(d, c, z);
    could optionally check that (a, b) w/y (c, d), but probably better to just handle via the post filter.

Indexes and subsort information:

  i := rawindex
  p := project(i, logicalindex);
  st := stepped(p, [a,b,c,d,e]);
  e := project(st, p2());
  f := compoundindexread;

Need to locate stepped
i) walk up to work out what is projected, and down.  Probably simplest done using a recursive function - should be relatively simple.  Don't merge with the index definition any more.
ii) Implement should be ok.  Have a flag to indicate if we spotted a STEPPED() identifier.  Complain if not a read.

Note:
  for search "a and b and date > x" it is much better to step (a,b) first before date because of condition complexity

  */