Big-Data-HPC-Platform/rtl/eclrtl/rtlkey.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 "jlib.hpp"
#include "jsort.hpp"
#include "jexcept.hpp"
#include "rtlkey.hpp"
#include "rtlkey2.hpp"
#include "eclrtl_imp.hpp"
#include "rtlrecord.hpp"
#include "rtlnewkey.hpp"

class CKeySegmentMonitor : implements IKeySegmentMonitor, public CInterface
{
protected:
    size32_t size;
    size32_t offset;
    unsigned fieldIdx;

public:
    IMPLEMENT_IINTERFACE;
    CKeySegmentMonitor(unsigned _fieldIdx, unsigned _offset, unsigned _size);

    virtual bool matchesBuffer(const void * rawRow) const override = 0;
    virtual bool matches(const RtlRow * rawRow) const override
    {
        return matchesBuffer(rawRow->queryRow());
    }

    virtual bool increment(void *keyval) const override;
    virtual unsigned queryFieldIndex() const override { return fieldIdx; }
    virtual unsigned getOffset() const override { return offset; }
    virtual unsigned getSize() const override { return size; }
    virtual bool isWild() const override { return false; }
    virtual bool isEmpty() const override { return false; }
    virtual bool isSigned() const override { return false; }
    virtual bool isLittleEndian() const override { return false; }
    virtual unsigned numFieldsRequired() const override { return 0; }  // Should rename to queryFieldIdx or similar

    virtual int docompare(const void * l, const void * r) const override
    {
        char *lptr = ((char *) l) + offset;
        char *rptr = ((char *) r) + offset;
        return memcmp(lptr, rptr, size);
    }

    virtual bool getBloomHash(hash64_t &hash) const override
    {
        return false;
    }

    virtual void setHigh(void *keyval) const override;

    virtual void copy(void * l, const void * r) const override
    {
        char *lptr = ((char *) l) + offset;
        char *rptr = ((char *) r) + offset;
        memcpy(lptr, rptr, size);
    }
};

class CWildKeySegmentMonitor : public CKeySegmentMonitor
{
public:
    CWildKeySegmentMonitor(unsigned _fieldIdx, unsigned _offset, unsigned _size);
    virtual bool matchesBuffer(const void *keyval) const override;
    virtual int docompare(const void *,const void *) const override;
    virtual void setLow(void *keyval) const override;
    virtual void endRange(void *keyval) const override;
    virtual bool isWild() const override { return true; }
    virtual bool isSimple() const override { return true; }
    virtual bool isWellKeyed() const override { return false; }
    virtual bool isOptional() const override { return true; }
    virtual StringBuffer &describe(StringBuffer &out, const RtlTypeInfo &type) const override { return out.append('*'); }
};

class CSetKeySegmentMonitor : public CKeySegmentMonitor
{
private:
    Owned<IStringSet> set;
    bool optional;

public:
    CSetKeySegmentMonitor(bool _optional, IStringSet *set, unsigned _fieldIdx, unsigned _offset, unsigned _size);

// IKeySegmentMonitor
    virtual bool increment(void *keyval) const override;
    virtual void setLow(void *keyval) const override;
    virtual bool matchesBuffer(const void *keyval) const override;
    virtual void endRange(void *keyval) const override;
    virtual bool isEmpty() const override { return set->isEmptySet(); }
    virtual bool isWellKeyed() const override;
    virtual bool isOptional() const override { return optional; }
    virtual bool isSimple() const override { return true; }
    virtual bool isSigned() const override { return set->isSigned(); }
    virtual bool isLittleEndian() const override { return !set->isBigEndian(); }

    virtual int docompare(const void * l, const void * r) const override
    {
        char *lptr = ((char *) l) + offset;
        char *rptr = ((char *) r) + offset;
        return set->memcmp(lptr, rptr, size);
    }
    virtual StringBuffer &describe(StringBuffer &out, const RtlTypeInfo &type) const override { return out.append("[...]"); }  // MORE - could do better
};

CKeySegmentMonitor::CKeySegmentMonitor(unsigned _fieldIdx, unsigned _offset, unsigned _size)
{
    size = _size;
    offset = _offset;
    fieldIdx = _fieldIdx;
}

bool CKeySegmentMonitor::increment(void *bufptr) const
{
    char *ptr = ((char *) bufptr) + offset;
    int i = size;
    while (i--)
    {
        ptr[i]++;
        if (ptr[i]!=0)
            return true;
    }
    return false;
}

void CKeySegmentMonitor::setHigh(void *bufptr) const
{ 
    // NOTE - effectively whenever this is called we are treating the segmonitor as if it was a wild one
    char *ptr = ((char *) bufptr) + offset;
    memset(ptr, 0xff, size); 
}


CWildKeySegmentMonitor::CWildKeySegmentMonitor(unsigned _fieldIdx, unsigned _offset, unsigned _size)
    : CKeySegmentMonitor(_fieldIdx, _offset, _size)
{
}

bool CWildKeySegmentMonitor::matchesBuffer(const void *keyval) const
{ 
    return true;
}

int CWildKeySegmentMonitor::docompare(const void *l, const void *r) const
{
    return 0;
}

void CWildKeySegmentMonitor::setLow(void *bufptr) const
{ 
    char *ptr = ((char *) bufptr) + offset;
    memset(ptr, 0, size); 
}

void CWildKeySegmentMonitor::endRange(void *bufptr) const
{ 
    char *ptr = ((char *) bufptr) + offset;
    memset(ptr, 0xff, size); 
}

CSetKeySegmentMonitor::CSetKeySegmentMonitor(bool _optional, IStringSet *_set, unsigned _fieldIdx, unsigned _offset, unsigned _size)
    : CKeySegmentMonitor(_fieldIdx, _offset, _size), set(_set)
{
    optional = _optional;
}

bool CSetKeySegmentMonitor::increment(void *bufptr) const
{
    char *ptr = ((char *) bufptr) + offset;
    bool ok = set->increment(ptr);
    if (ok)
    {
        unsigned nextTransition;
        bool res = set->inRange(ptr, nextTransition);
        if (!res)
        {
            if ((unsigned) -1 == nextTransition) return false;
            set->getTransitionValue(ptr, nextTransition);
        }
    }
    return ok;
}

void CSetKeySegmentMonitor::setLow(void *bufptr) const
{
    char *ptr = ((char *) bufptr) + offset;
    if (set->transitionCount())
        set->getTransitionValue(ptr, 0);
    else
        memset(ptr, 0, size); // MORE - should really trap earlier
}

void CSetKeySegmentMonitor::endRange(void *bufptr) const
{
    char *ptr = ((char *) bufptr) + offset;
    unsigned nextTransition;
    bool res = set->inRange(ptr, nextTransition);
    assertex(res);
    verifyex(set->getTransitionValue(ptr, nextTransition));
}

bool CSetKeySegmentMonitor::matchesBuffer(const void *bufptr) const
{
    char *ptr = ((char *) bufptr) + offset;
    return set->inRange(ptr);
}

bool CSetKeySegmentMonitor::isWellKeyed() const
{
    // This check determines whether or not keyed, opt considers this field to be keyed.
    // The goal is to allow sets but not ranges, slightly complicated by the fact that adjacent values in a set turn into ranges.
    return set->numValues() < 50;
}

class CSingleKeySegmentMonitorBase : public CKeySegmentMonitor
{
protected:
    void *val;
    bool optional;
public:
    CSingleKeySegmentMonitorBase(bool _optional, const void *_val, unsigned _fieldIdx, unsigned _offset, unsigned _size)
        : CKeySegmentMonitor(_fieldIdx, _offset, _size)
    {
        if (_val)
        {
            val = malloc(_size);
            memcpy(val, _val, _size);
        }
        else
            val = NULL;
        optional = _optional;
    }
    CSingleKeySegmentMonitorBase(bool _optional, unsigned _fieldIdx, unsigned _offset, unsigned _size)
        : CKeySegmentMonitor(_fieldIdx, _offset, _size)
    {
        val = NULL;
        optional = _optional;
    }

    ~CSingleKeySegmentMonitorBase()
    {
        free(val);
    }

// IKeySegmentMonitor
    virtual bool increment(void *bufptr) const override
    {
        // Set to next permitted value above current
        if (docompare(bufptr, ((char *) val)-offset) < 0)  // NOTE - we subtract offset from val before passing because the compare will add it on again. Confusing or what...
        {
            char *ptr = ((char *) bufptr) + offset;
            memcpy(ptr, val, size);
            return true;
        }
        else
            return false;
    }
    virtual void setLow(void *bufptr) const override
    {
        // Set to lowest permitted value
        char *ptr = ((char *) bufptr) + offset;
        memcpy(ptr, val, size);
    }
    virtual bool matchesBuffer(const void *bufptr) const override
    {
        // Is current a permitted value?
        char *ptr = ((char *) bufptr) + offset;
        return memcmp(ptr, val, size) == 0;
    }

    virtual void endRange(void *bufptr) const override
    {
        // Set to last permitted value in the range that includes current (which is asserted to be valid)
        dbgassertex(matchesBuffer(bufptr));
    }

    virtual bool isWellKeyed() const override { return true; }
    virtual bool isOptional() const override { return optional; }
    virtual bool isSimple() const override { return true; }
    virtual StringBuffer &describe(StringBuffer &out, const RtlTypeInfo &type) const override
    {
        size32_t size;
        rtlDataAttr text;
        type.getString(size, text.refstr(), val);
        if (type.isNumeric())
            return out.append(size, text.getstr());
        else
            return out.appendf("'%*s'", size, text.getstr());
    }
};

class CSingleKeySegmentMonitor : public CSingleKeySegmentMonitorBase
{
public:
    CSingleKeySegmentMonitor(bool _optional, const void *_val, unsigned _fieldIdx, unsigned _offset, unsigned _size)
        : CSingleKeySegmentMonitorBase(_optional, _val, _fieldIdx, _offset, _size)
    {
    }

    virtual bool isSigned() const override { return false; }
    virtual bool isLittleEndian() const override { return false; }
    virtual bool getBloomHash(hash64_t &hash) const override
    {
        if (!val)
            return false;
        hash = rtlHash64Data(size, val, hash);
        return true;
    }
};

class CSingleBigSignedKeySegmentMonitor : public CSingleKeySegmentMonitorBase
{
public:
    CSingleBigSignedKeySegmentMonitor(bool _optional, const void *_val, unsigned _fieldIdx, unsigned _offset, unsigned _size)
        : CSingleKeySegmentMonitorBase(_optional, _val, _fieldIdx, _offset, _size)
    {
    }

    virtual int docompare(const void *l, const void *r) const override
    {
        return memcmpbigsigned(((char *) l) + offset, ((char *) r) + offset, size);
    }

    virtual bool isSigned() const override { return true; }
    virtual bool isLittleEndian() const override { return false; }
};

class CSingleLittleSignedKeySegmentMonitor : public CSingleKeySegmentMonitorBase
{
public:
    CSingleLittleSignedKeySegmentMonitor(bool _optional, const void *_val, unsigned _fieldIdx, unsigned _offset, unsigned _size)
        : CSingleKeySegmentMonitorBase(_optional, _val, _fieldIdx, _offset, _size)
    {
    }

    virtual int docompare(const void *l, const void *r) const
    {
        return memcmplittlesigned(((char *) l) + offset, ((char *) r) + offset, size);
    }

    virtual bool isSigned() const override { return true; }
    virtual bool isLittleEndian() const override { return true; }
};

class CSingleLittleKeySegmentMonitor : public CSingleKeySegmentMonitorBase
{
public:
    CSingleLittleKeySegmentMonitor(bool _optional, const void *_val, unsigned _fieldIdx, unsigned _offset, unsigned _size)
        : CSingleKeySegmentMonitorBase(_optional, _val, _fieldIdx, _offset, _size)
    {
    }

    virtual int docompare(const void *l, const void *r) const override
    {
        return memcmplittleunsigned(((char *) l) + offset, ((char *) r) + offset, size);
    }

    virtual bool isSigned() const override { return false; }
    virtual bool isLittleEndian() const override { return true; }
};

ECLRTL_API IStringSet *createRtlStringSet(size32_t size)
{
    return createStringSet(size);
}

ECLRTL_API IStringSet *createRtlStringSetEx(size32_t size, bool bigEndian, bool isSigned)
{
    return createStringSet(size, bigEndian, isSigned);
}

ECLRTL_API IStringSet * rtlUnionSet(IStringSet * lhs, IStringSet * rhs)
{
    if (lhs->isEmptySet())
        return LINK(rhs);
    else if (lhs->isFullSet())
        return LINK(lhs);

    if (rhs->isEmptySet())
        return LINK(lhs);
    else if (rhs->isFullSet())
        return LINK(rhs);

    return lhs->unionSet(rhs);
}

ECLRTL_API IStringSet * rtlIntersectSet(IStringSet * lhs, IStringSet * rhs)
{
    if (lhs->isFullSet())
        return LINK(rhs);
    else if (lhs->isEmptySet())
        return LINK(lhs);

    if (rhs->isFullSet())
        return LINK(lhs);
    else if (rhs->isEmptySet())
        return LINK(rhs);

    return lhs->intersectSet(rhs);
}


IKeySegmentMonitor *createKeySegmentMonitor(bool optional, IStringSet *set, unsigned _fieldIdx, unsigned _offset, unsigned _size)
{
    if (!set)
        return new CWildKeySegmentMonitor(_fieldIdx, _offset, _size);

    Owned<IStringSet> removeSet = set; // make sure set is released if optimized out.
    if (set->isSingleValue())
    {
        void *data = alloca(_size);
        set->getTransitionValue(data, 0);
        if (set->isSigned())
        {
            if (set->isBigEndian())
                return createSingleBigSignedKeySegmentMonitor(optional, _fieldIdx, _offset, _size, data);
            else
                return createSingleLittleSignedKeySegmentMonitor(optional, _fieldIdx, _offset, _size, data);
        }
        else
        {
            if (set->isBigEndian())
                return createSingleKeySegmentMonitor(optional, _fieldIdx, _offset, _size, data);
            else
                return createSingleLittleKeySegmentMonitor(optional, _fieldIdx, _offset, _size, data);
        }
    }
    else if (set->isFullSet())
        return new CWildKeySegmentMonitor(_fieldIdx, _offset, _size);
    else
        return new CSetKeySegmentMonitor(optional, removeSet.getClear(), _fieldIdx, _offset, _size);
}

ECLRTL_API IStringSet *createRtlStringValue(size32_t size, const char * value)
{
    IStringSet * set = createStringSet(size);
    set->addRange(value, value);
    return set;
}

IKeySegmentMonitor *createWildKeySegmentMonitor(unsigned _fieldIdx, unsigned _offset, unsigned _size)
{
    return new CWildKeySegmentMonitor(_fieldIdx, _offset, _size);
}

IKeySegmentMonitor *createEmptyKeySegmentMonitor(bool optional, unsigned _fieldIdx, unsigned _offset, unsigned _size)
{
    return new CSetKeySegmentMonitor(optional, createStringSet(_size), _fieldIdx, _offset, _size);
}

ECLRTL_API IKeySegmentMonitor *createSingleKeySegmentMonitor(bool optional, unsigned _fieldIdx, unsigned offset, unsigned size, const void * value)
{
    return new CSingleKeySegmentMonitor(optional, value, _fieldIdx, offset, size);
}

ECLRTL_API IKeySegmentMonitor *createSingleBigSignedKeySegmentMonitor(bool optional, unsigned fieldIdx, unsigned offset, unsigned size, const void * value)
{
    return new CSingleBigSignedKeySegmentMonitor(optional, value, fieldIdx, offset, size);
}

ECLRTL_API IKeySegmentMonitor *createSingleLittleSignedKeySegmentMonitor(bool optional, unsigned fieldIdx, unsigned offset, unsigned size, const void * value)
{
    // MORE - common int sizes 1,2,4 (8?) might be better done with dedicated subclasses
    return new CSingleLittleSignedKeySegmentMonitor(optional, value, fieldIdx, offset, size);
}

ECLRTL_API IKeySegmentMonitor *createSingleLittleKeySegmentMonitor(bool optional, unsigned fieldIdx, unsigned offset, unsigned size, const void * value)
{
    // MORE - common int sizes 1,2,4 (8?) might be better done with dedicated subclasses
    return new CSingleLittleKeySegmentMonitor(optional, value, fieldIdx, offset, size);
}

ECLRTL_API IKeySegmentMonitor *createDummyKeySegmentMonitor(unsigned _fieldIdx, unsigned _offset, unsigned _size, bool isSigned, bool isLittleEndian)
{
    if (isSigned)
        if (isLittleEndian)
            return new CSingleLittleSignedKeySegmentMonitor(false, NULL, _fieldIdx, _offset, _size);
        else
            return new CSingleBigSignedKeySegmentMonitor(false, NULL, _fieldIdx, _offset, _size);
    else
        if (isLittleEndian)
            return new CSingleLittleKeySegmentMonitor(false, NULL, _fieldIdx, _offset, _size);
        else
            return new CSingleKeySegmentMonitor(false, NULL, _fieldIdx, _offset, _size);
        }

ECLRTL_API int memcmpbigsigned(const void *l, const void *r, unsigned size)
{
    signed int diff = ((signed char *) l)[0]-((signed char *) r)[0];
    if (diff)
        return diff;
    for(unsigned i = 1; i < size; i++)
    {
        diff = ((unsigned char *) l)[i]-((unsigned char *) r)[i];
        if (diff)
            return diff;
    }
    return 0;
}

ECLRTL_API int memcmplittleunsigned(const void *l, const void *r, unsigned size)
{
    while (size)
    {
        size--;
        int diff = ((unsigned char *) l)[size]-((unsigned char *) r)[size];
        if (diff)
            return diff;
    }
    return 0;
}

ECLRTL_API int memcmplittlesigned(const void *l, const void *r, unsigned size)
{
    size--;
    signed int diff = ((signed char *) l)[size]-((signed char *) r)[size];
    if (diff)
        return diff;
    while (size)
    {
        size--;
        diff = ((unsigned char *) l)[size]-((unsigned char *) r)[size];
        if (diff)
            return diff;
    }
    return 0;
}

class CStringSet : implements IStringSet, public CInterface
{
protected:
    size32_t size;
    IArrayOf<ITransition> transitions;

    IStringSet *unionOrIntersect(IStringSet *r, bool isUnion);

    virtual CStringSet *createEmptySet() = 0;
    virtual bool decrement(void *val) const = 0;
    virtual bool increment(void *val) const = 0;
    virtual int memcmp(const void *val1, const void *val2, size32_t size) const = 0;
    virtual unsigned getCardinality(const void *val1, const void *val2, size32_t size) const = 0;
    virtual void memset(void *ptr, int val, size32_t size) const = 0;
    virtual bool isLowVal(const void *val) const = 0;
    virtual bool isHighVal(const void *val) const = 0;
    bool oneless(const void *l, const void *r) const;

    void addTransitionAt(const void *val, bool state, unsigned pos);
    void appendTransition(ITransition *t);

public:
    IMPLEMENT_IINTERFACE;
    CStringSet(size32_t size);

// IStringSet
    virtual void addRange(const void *loval, const void *hival);
    virtual void addAll();
    virtual ITransition *queryTransition(unsigned idx);
    virtual bool getTransitionValue(void *value, unsigned idx);
    virtual void killRange(const void *loval, const void *hival);
    virtual bool inRange(const void *val) const;
    virtual bool inRange(const void *val, unsigned &transition) const;
    virtual size32_t getSize() { return size; };
    virtual void reset();
    virtual unsigned transitionCount();
    virtual IStringSet *invertSet();
    virtual IStringSet *unionSet(IStringSet *);
    virtual IStringSet *intersectSet(IStringSet *);
    virtual const char *describe(StringBuffer &ret);
    virtual bool isEmptySet() const { return transitions.length()==0; }
    virtual bool isFullSet() const 
    { 
        return transitions.length()==2 && 
            isLowVal(transitions.item(0).getValue()) &&
            isHighVal(transitions.item(1).getValue());
    }
    virtual bool isSingleValue() const
    {
        return transitions.length()==2 && 
            memcmp(transitions.item(0).getValue(), transitions.item(1).getValue(), size) == 0;
    }

    virtual unsigned numValues() const
    {
        unsigned ret = 0;
        unsigned idx = 0;
        while (transitions.isItem(idx+1))
        {
            unsigned thisrange = getCardinality(transitions.item(idx).getValue(), transitions.item(idx+1).getValue(), size);
            if (thisrange + ret < ret)
                return (unsigned) -1;
            ret += thisrange;
            idx += 2;
        }
        return ret;
    }
};

class CBigUnsignedStringSet : public CStringSet
{
protected:
    virtual CStringSet *createEmptySet()
    {
        return new CBigUnsignedStringSet(size);
    }

    virtual bool increment(void *_val) const
    {
        unsigned char *val = (unsigned char *)_val;
        int i = size;
        while (i--)
        {
            val[i]++;
            if (val[i]!=0)
                return true;
        }
        return false;
    }

    virtual bool decrement(void *_val) const
    {
        unsigned char *val = (unsigned char *)_val;
        int i = size;
        while (i--)
        {
            val[i]--;
            if ((unsigned char)val[i]!=0xff)
                return true;
        }
        return false;
    }

    virtual int memcmp(const void *val1, const void *val2, size32_t size) const
    {
        return ::memcmp(val1, val2, size);
    }

    virtual void memset(void *ptr, int val, size32_t size) const
    {
        ::memset(ptr, val, size);
    }

    virtual unsigned getCardinality(const void *val1, const void *val2, size32_t size) const
    {
        unsigned char *p1 = (unsigned char *) val1;
        unsigned char *p2 = (unsigned char *) val2;
        unsigned ret = 1;
        unsigned mult = 1;
        while (size--)
        {
            unsigned diff = p2[size] - p1[size];
            if (diff)
            {
                if (!mult) 
                    return (unsigned) -1;
                else
                    ret += diff * mult;
            }
            if (mult*256 < mult)
                mult = 0;
            else
                mult *= 256;
        }
        return ret;
    }

    virtual bool isHighVal(const void *val) const
    {
        const unsigned char *vval = (const unsigned char *) val;
        for (unsigned i = 0; i < size; i++)
            if (vval[i] != 0xff)
                return false;
        return true;
    }

    virtual bool isLowVal(const void *val) const
    {
        const unsigned char *vval = (const unsigned char *) val;
        for (unsigned i = 0; i < size; i++)
            if (vval[i] != 0x00)
                return false;
        return true;
    }

    virtual bool isSigned() const { return false; }
    virtual bool isBigEndian() const { return true; }

public:
    CBigUnsignedStringSet(unsigned size) : CStringSet(size) {}
};

class CBigSignedStringSet : public CBigUnsignedStringSet
{
protected:
    virtual CStringSet *createEmptySet()
    {
        return new CBigSignedStringSet(size);
    }

    // increment and decrement are same as unsigned

    virtual int memcmp(const void *val1, const void *val2, size32_t size) const
    {
        return ::memcmpbigsigned(val1, val2, size);
    }

    virtual void memset(void *ptr, int val, size32_t size) const
    {
        ::memset(ptr, val, size);
        switch(val)
        {
        case 0:
            *(unsigned char *) ptr = 0x80;
            break;
        case 0xff:
            *(unsigned char *) ptr = 0x7f;
            break;
        default:
            throwUnexpected();
        }
    }

    virtual bool isHighVal(const void *val) const
    {
        const unsigned char *vval = (const unsigned char *) val;
        if (vval[0] != 0x7f)
            return false;
        for (unsigned i = 1; i < size; i++)
            if (vval[i] != 0xff)
                return false;
        return true;
    }

    virtual bool isLowVal(const void *val) const
    {
        const unsigned char *vval = (const unsigned char *) val;
        if (vval[0] != 0x80)
            return false;
        for (unsigned i = 1; i < size; i++)
            if (vval[i] != 0x00)
                return false;
        return true;
    }

    virtual bool isSigned() const { return true; }
    virtual bool isBigEndian() const { return true; }

public:
    CBigSignedStringSet(unsigned size) : CBigUnsignedStringSet(size) {}
};

class CLittleUnsignedStringSet : public CStringSet
{
protected:
    virtual CStringSet *createEmptySet()
    {
        return new CLittleUnsignedStringSet(size);
    }

    virtual bool increment(void *_val) const
    {
        unsigned char *val = (unsigned char *)_val;
        unsigned i = 0;
        while (i < size)
        {
            val[i]++;
            if (val[i]!=0)
                return true;
            i++;
        }
        return false;
    }

    virtual unsigned getCardinality(const void *val1, const void *val2, size32_t size) const
    {
        unsigned char *p1 = (unsigned char *) val1;
        unsigned char *p2 = (unsigned char *) val2;
        unsigned ret = 1;
        unsigned mult = 1;
        unsigned i = 0;
        while (i < size)
        {
            unsigned diff = p2[i] - p1[i];
            if (diff)
            {
                if (!mult) 
                    return (unsigned) -1;
                else
                    ret += diff * mult;
            }
            if (mult*256 < mult)
                mult = 0;
            else
                mult *= 256;
            i++;
        }
        return ret;
    }

    virtual bool decrement(void *_val) const
    {
        unsigned char *val = (unsigned char *)_val;
        unsigned i = 0;
        while (i < size)
        {
            val[i]--;
            if ((unsigned char)val[i]!=0xff)
                return true;
            i++;
        }
        return false;
    }

    virtual int memcmp(const void *val1, const void *val2, size32_t size) const
    {
        return ::memcmplittleunsigned(val1, val2, size);
    }

    virtual void memset(void *ptr, int val, size32_t size) const
    {
        ::memset(ptr, val, size);
    }

    virtual bool isHighVal(const void *val) const
    {
        const unsigned char *vval = (const unsigned char *) val;
        for (unsigned i = 0; i < size; i++)
            if (vval[i] != 0xff)
                return false;
        return true;
    }

    virtual bool isLowVal(const void *val) const
    {
        const unsigned char *vval = (const unsigned char *) val;
        for (unsigned i = 0; i < size; i++)
            if (vval[i] != 0x00)
                return false;
        return true;
    }

    virtual bool isSigned() const { return false; }
    virtual bool isBigEndian() const { return false; }
public:
    CLittleUnsignedStringSet(unsigned size) : CStringSet(size) {}
};

class CLittleSignedStringSet : public CLittleUnsignedStringSet
{
protected:
    virtual CStringSet *createEmptySet()
    {
        return new CLittleSignedStringSet(size);
    }

    // increment and decrement are same as unsigned

    virtual int memcmp(const void *val1, const void *val2, size32_t size) const
    {
        return ::memcmplittlesigned(val1, val2, size);
    }

    virtual void memset(void *ptr, int val, size32_t size) const
    {
        if (size > 1)
            ::memset(ptr, val, size);
        unsigned char *pptr = (unsigned char *) ptr;
        switch(val)
        {
        case 0:
            pptr[size-1] = 0x80;
            break;
        case 0xff:
            pptr[size-1] = 0x7f;
            break;
        default:
            throwUnexpected();
        }
    }

    virtual bool isHighVal(const void *val) const
    {
        const unsigned char *vval = (const unsigned char *) val;
        if (vval[size-1] != 0x7f)
            return false;
        for (unsigned i = 0; i < size-1; i++)
            if (vval[i] != 0xff)
                return false;
        return true;
    }

    virtual bool isLowVal(const void *val) const
    {
        const unsigned char *vval = (const unsigned char *) val;
        if (vval[size-1] != 0x80)
            return false;
        for (unsigned i = 0; i < size-1; i++)
            if (vval[i] != 0x00)
                return false;
        return true;
    }

    virtual bool isSigned() const { return true; }
    virtual bool isBigEndian() const { return false; }
public:
    CLittleSignedStringSet(unsigned size) : CLittleUnsignedStringSet(size) {}
};


class CTransition : implements ITransition, public CInterface
{
private:
    bool state;                 // note: should move before ITransition to pack better in 64bit
    const void *val;

public:
    IMPLEMENT_IINTERFACE;

    CTransition(const void *_val, bool _state)
    {
        val = _val;
        state = _state;
    }
    ~CTransition() { free((void *) val); }

// ITransition

    virtual bool getState() const override { return state; }
    virtual const void *getValue() const override { return val; }
};

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

CStringSet::CStringSet(size32_t _size)
{
    size = _size;
}

void CStringSet::reset()
{
    transitions.kill();
}

bool CStringSet::oneless(const void *l, const void *r) const
{
    // MORE - would be more efficient to make this virtual like the memcmp...
    void *t = alloca(size);
    memcpy(t, r, size);
    decrement(t);
    return memcmp(l, t, size)==0;
}

unsigned CStringSet::transitionCount()
{
    return transitions.ordinality();
}

void CStringSet::addTransitionAt(const void *val, bool state, unsigned pos)
{
    void *newval = malloc(size);
    memcpy(newval, val, size);
    transitions.add(* new CTransition(newval, state), pos);
}

void CStringSet::appendTransition(ITransition *t)
{
    if (t->getState() && transitions.length())
    {
        unsigned lastidx = transitions.length()-1;
        ITransition &prev = transitions.item(lastidx);
        assertex(prev.getState()==!t->getState());
        if (oneless(prev.getValue(), t->getValue()))
        {
            transitions.remove(lastidx);
            t->Release();
            return;
        }
    }
    transitions.append(*t);
}

void CStringSet::addRange(const void *loval, const void *hival)
{
    if (!loval)
    {
        void *x = alloca(size);
        memset(x, 0, size);
        loval = x;
    }
    if (!hival)
    {
        void *x = alloca(size);
        memset(x, 0xff, size);
        hival = x;
    }
    if (memcmp(loval, hival, size) > 0)
        return;

    unsigned idx;
    bool inset = false;
    int b = transitions.ordinality();
    if (!b)
    {
        addTransitionAt(loval, true, 0);
        addTransitionAt(hival, false, 1);
        return;
    }
    else
    {
        // binchop to find last transition > val...
        unsigned int a = 0;
        int rc;
        while ((int)a<b)
        {
            int i = a+(b+1-a)/2;
            rc = memcmp(loval, transitions.item(i-1).getValue(), size);
            if (rc>0)
                a = i;
            else
                b = i-1;
        }
        if (a>0)
        {
            idx = a;
            ITransition &t = transitions.item(idx-1);
            if(!t.getState())
            {
                if (oneless(t.getValue(), loval)) 
                    transitions.remove(--idx);
                else
                    addTransitionAt(loval, true, idx++);
            }
            else
                inset = true;
        }
        else
        {
            addTransitionAt(loval, true, 0);
            idx = 1;
        }
    }

    while (transitions.isItem(idx))
    {
        ITransition &t = transitions.item(idx);
        int diff = memcmp(t.getValue(), hival, size);
        if (diff <= 0)
        {
            inset = t.getState();
            transitions.remove(idx);
        }
        else
            break;
    }

    if (!inset)
    {
        if (transitions.isItem(idx))
        {
            ITransition &t = transitions.item(idx);
            assertex(t.getState());
            if (oneless(hival, t.getValue()))
            {
                transitions.remove(idx);
                return;
            }
        }
        addTransitionAt(hival, false, idx);
    }
}

void CStringSet::killRange(const void *loval, const void *hival)
{
    if (!loval)
    {
        void *x = alloca(size);
        memset(x, 0, size);
        loval = x;
    }
    if (!hival)
    {
        void *x = alloca(size);
        memset(x, 0xff, size);
        hival = x;
    }
    assertex(memcmp(loval, hival, size) <= 0);
    bool inset = false;
    ForEachItemIn(idx, transitions)
    {
        ITransition &t = transitions.item(idx);
        int diff = memcmp(t.getValue(), loval, size);
        if (diff < 0)
            inset = t.getState();
        else
            break;
    }
    if (inset)
    {
        void *nlo = alloca(size);
        memcpy(nlo, loval, size);
        decrement(nlo);
        addTransitionAt(nlo, false, idx++);
    }
    while (transitions.isItem(idx))
    {
        ITransition &t = transitions.item(idx);
        int diff = memcmp(t.getValue(), hival, size);
        if (diff <= 0)
        {
            inset = t.getState();
            transitions.remove(idx);
        }
        else
            break;
    }
    if (inset)
    {
        void *nhi = alloca(size);
        memcpy(nhi, hival, size);
        increment(nhi);
        addTransitionAt(nhi, true, idx);
    }
}

void CStringSet::addAll()
{
    reset();
    void *val = alloca(size);
    memset(val, 0, size);
    addTransitionAt(val, true, 0);
    memset(val, 0xff, size);
    addTransitionAt(val, false, 1);
}

const char *CStringSet::describe(StringBuffer &ret)
{
    ret.append('[');
    ForEachItemIn(idx, transitions)
    {
        ITransition &t = transitions.item(idx);
        if (t.getState())
        {
            if (idx)
                ret.append(',');
        }
        else
            ret.append("..");
        appendURL(&ret, (char *) t.getValue(), size, true);
    }
    ret.append(']');
    return ret.str();
}

bool CStringSet::inRange(const void *val) const
{
    unsigned nextTransition;
    return inRange(val, nextTransition);
}

bool CStringSet::inRange(const void *val, unsigned &nextTransition) const
{
    int b = transitions.ordinality();
    if (!b)
    {
        nextTransition = (unsigned) -1; 
        return false; 
    }
    else if (b >= 4)
    {
        // binchop to find last transition >= val...
        unsigned int a = 0;
        int rc;
        while ((int)a<b)
        {
            int i = a+(b+1-a)/2;
            rc = memcmp(val, transitions.item(i-1).getValue(), size);
            if (rc>=0)
                a = i;
            else
                b = i-1;
        }
        if (a>0)
        {
            nextTransition = (a>=transitions.ordinality())? (unsigned) -1: a; // a is first transition that is > val
            a--;
            if (transitions.item(a).getState())
                return true;
            if (memcmp(val, transitions.item(a).getValue(), size)==0)
            {
                nextTransition = a;
                return true;
            }
            return false;

        }
        else
        {
            nextTransition = 0;
            return false;
        }
    }
    else
    {
        bool inset = false;
        ForEachItemIn(idx, transitions)
        {
            ITransition &t = transitions.item(idx);
            int diff = memcmp(t.getValue(), val, size);
            if (t.getState())
            {
                if (diff <= 0)
                    inset = true;
                if (diff == 0)
                {
                    idx++;
                    break;
                }
                else if (diff > 0)
                    break;
            }
            else
            {
                if (diff >= 0)
                    break;
                if (diff < 0)
                    inset = false;
            }
        }
        nextTransition = (idx>=transitions.ordinality())? (unsigned) -1: idx;
        return inset;
    }
}

IStringSet *CStringSet::unionOrIntersect(IStringSet *r, bool isUnion)
{
    bool inA = false;
    bool inB = false;
    bool state = false;
    assertex(r->getSize()==size);
    int idxA = 0;
    int idxB = 0;
    ITransition *tA = queryTransition(idxA);
    ITransition *tB = r->queryTransition(idxB);
    CStringSet *result = createEmptySet();
    for (;;)
    {
        int diff;
        if (tA == NULL)
        {
            if (tB == NULL)
                break;
            else
                diff = 1;
        }
        else if (tB == NULL)
            diff = -1;
        else
            diff = memcmp(tA->getValue(), tB->getValue(), size);
        ITransition *t = NULL;
        if (!diff)
        {
            diff = (int) tB->getState() - (int) tA->getState(); // leading edge sorts before trailing edge for intersect...
            if (isUnion)
                diff = -diff;                                   // trailing edge sorts before leading edge for union...
        }
        if (diff <= 0)
        {
            inA = tA->getState();
            t = tA;
            idxA++;
            tA = queryTransition(idxA);
        }
        if (diff >= 0)
        {
            inB = tB->getState();
            t = tB;
            idxB++;
            tB = r->queryTransition(idxB);
        }
        bool newState;
        if (isUnion)
            newState = inA || inB;
        else
            newState = inA && inB;
        if (newState != state)
        {
            state = newState;
            t->Link();
            result->appendTransition(t);
        }
    }
    return result;
}

IStringSet *CStringSet::invertSet()
{
    CStringSet *result = createEmptySet();
    result->addAll();
    bool inset = false;
    void *loval = alloca(size);
    void *hival = alloca(size);
    memset(loval, 0, size);
    ForEachItemIn(idx, transitions)
    {
        ITransition &t = transitions.item(idx);
        assertex(t.getState() == !inset);
        if (inset)
        {
            memcpy(hival, t.getValue(), size);
            result->killRange(loval, hival);
        }
        else
            memcpy(loval, t.getValue(), size);
        inset = t.getState();
    }
    if (inset)
    {
        memset(hival, 0xff, size);
        result->killRange(loval, hival);
    }
    return result;
}

IStringSet *CStringSet::unionSet(IStringSet *other)
{
    return unionOrIntersect(other, true);
}

IStringSet *CStringSet::intersectSet(IStringSet *other)
{
    return unionOrIntersect(other, false);
}

ITransition *CStringSet::queryTransition(unsigned int idx)
{
    if (transitions.isItem(idx)) 
    {
        ITransition *t = &transitions.item(idx);
        return t;
    }
    else
        return NULL;
}

bool CStringSet::getTransitionValue(void *value, unsigned int idx)
{
    if (idx == (unsigned) -1 || idx >= transitions.ordinality()) return false;
    ITransition &t = transitions.item(idx);
    memcpy(value, t.getValue(), size);
    return true;
}

IStringSet *createStringSet(size32_t size)
{
    return new CBigUnsignedStringSet(size);
}

IStringSet *createStringSet(size32_t size, bool bigEndian, bool isSigned)
{
    if (bigEndian)
    {
        if (isSigned)
            return new CBigSignedStringSet(size);
        else
            return new CBigUnsignedStringSet(size);
    }
    else
    {
        if (isSigned)
            return new CLittleSignedStringSet(size);
        else
            return new CLittleUnsignedStringSet(size);
    }
}

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

class LegacySetCreator : implements ISetCreator
{
public:
    LegacySetCreator(IStringSet & _set, size32_t _minRecordSize, const RtlTypeInfo * _fieldType)
    : set(_set), fieldType(_fieldType), minRecordSize(_minRecordSize) {}

    virtual void addRange(TransitionMask lowerMask, const StringBuffer & lowerString, TransitionMask upperMask, const StringBuffer & upperString) override
    {
        MemoryBufferBuilder lobuilder(lobuffer.clear(), minRecordSize);
        fieldType->buildUtf8(lobuilder, 0, nullptr, lowerString.length(), lowerString.str());

        MemoryBufferBuilder hibuilder(hibuffer.clear(), minRecordSize);
        fieldType->buildUtf8(hibuilder, 0, nullptr, upperString.length(), upperString.str());

        set.addRange(lobuffer.toByteArray(), hibuffer.toByteArray());
        if (!(lowerMask & CMPeq))
            set.killRange(lobuffer.toByteArray(), lobuffer.toByteArray());
        if (!(upperMask & CMPeq))
            set.killRange(hibuffer.toByteArray(), hibuffer.toByteArray());
    }

protected:
    IStringSet & set;
    const RtlTypeInfo *fieldType;
    size32_t minRecordSize;
    MemoryBuffer lobuffer;
    MemoryBuffer hibuffer;
};

void deserializeSet(IStringSet & set, size32_t minRecordSize, const RtlTypeInfo * fieldType, const char * filter)
{
    LegacySetCreator creator(set, minRecordSize, fieldType);
    deserializeSet(creator, filter);
}

#ifdef _USE_CPPUNIT
#include <cppunit/extensions/HelperMacros.h>

/*
class IStdException : extends std::exception
{
    Owned<IException> jException;
public:
    IStdException(IException *E) : jException(E) {};
};
*/

class SegmentMonitorTest : public CppUnit::TestFixture  
{
    CPPUNIT_TEST_SUITE( SegmentMonitorTest  );
        CPPUNIT_TEST(testOptional);
    CPPUNIT_TEST_SUITE_END();

protected:

    void testOptional()
    {
        Owned<IKeySegmentMonitor> wild0_20 = createWildKeySegmentMonitor(0, 0, 20);
        Owned<IKeySegmentMonitor> wild10_10 = createWildKeySegmentMonitor(1, 10,10);
        Owned<IStringSet> abcdef = createStringSet(10);
        abcdef->addRange("ABCDEFGHIJ", "ABCDEFGHIJ");
        Owned<IKeySegmentMonitor> opt0_20 = createSingleKeySegmentMonitor(true, 0, 0,20, "abcdefghijklmnopqrst");
        Owned<IKeySegmentMonitor> opt20_10 = createKeySegmentMonitor(true, LINK(abcdef), 1, 20, 10);
        Owned<IKeySegmentMonitor> opt30_10 = createSingleKeySegmentMonitor(true, 2, 30, 10, "KLMNOPQRST");
        Owned<IKeySegmentMonitor> nonOpt0_10 = createSingleKeySegmentMonitor(false, 0, 0,10, "abcdefghij");
        Owned<IKeySegmentMonitor> nonOpt0_20 = createSingleKeySegmentMonitor(false, 0, 0,20, "abcdefghijklmnopqrst");
        Owned<IKeySegmentMonitor> nonOpt20_10 = createKeySegmentMonitor(false, LINK(abcdef), 1, 20, 10);
        Owned<IKeySegmentMonitor> nonOpt30_10 = createSingleKeySegmentMonitor(false, 2, 30, 10, "KLMNOPQRST");
        CPPUNIT_ASSERT(wild0_20->isOptional());
        CPPUNIT_ASSERT(opt20_10->isOptional());
        CPPUNIT_ASSERT(opt30_10->isOptional());
        CPPUNIT_ASSERT(!nonOpt0_10->isOptional());
        CPPUNIT_ASSERT(!nonOpt0_20->isOptional());
        CPPUNIT_ASSERT(!nonOpt20_10->isOptional());
        CPPUNIT_ASSERT(!nonOpt30_10->isOptional());

#if 0
        IKeySegmentMonitorArray segments;
        segments.append(*LINK(wild0_20));
        segments.append(*LINK(opt20_10));
        
        CPPUNIT_ASSERT(segments.ordinality() == 1);
        CPPUNIT_ASSERT(segments.item(0).isWild());
        CPPUNIT_ASSERT(segments.item(0).getOffset() == 0);
        CPPUNIT_ASSERT(segments.item(0).getSize() == 30);

        segments.kill();
        segments.append(*LINK(wild0_20));
        segments.append(*LINK(opt20_10));
        segments.append(*LINK(nonOpt30_10));
        CPPUNIT_ASSERT(segments.ordinality() == 2);
        CPPUNIT_ASSERT(segments.item(0).isWild());
        CPPUNIT_ASSERT(segments.item(0).getOffset() == 0);
        CPPUNIT_ASSERT(segments.item(0).getSize() == 30);
        CPPUNIT_ASSERT(!segments.item(1).isWild());
        CPPUNIT_ASSERT(segments.item(1).getOffset() == 30);
        CPPUNIT_ASSERT(segments.item(1).getSize() == 10);

        segments.kill();
        segments.append(*LINK(nonOpt0_20));
        segments.append(*LINK(opt20_10));
        segments.append(*LINK(nonOpt30_10));
        CPPUNIT_ASSERT(segments.ordinality() == 3);
        CPPUNIT_ASSERT(!segments.item(1).isWild());
        CPPUNIT_ASSERT(segments.item(1).getOffset() == 20);
        CPPUNIT_ASSERT(segments.item(1).getSize() == 10);

        segments.kill();
        segments.append(*LINK(nonOpt0_10));
        segments.append(*LINK(wild10_10));
        segments.append(*LINK(opt20_10));
        segments.append(*LINK(nonOpt30_10));
        CPPUNIT_ASSERT(segments.ordinality() == 3);
        CPPUNIT_ASSERT(!segments.item(0).isWild());
        CPPUNIT_ASSERT(segments.item(1).isWild());
        CPPUNIT_ASSERT(segments.item(1).getOffset() == 10);
        CPPUNIT_ASSERT(segments.item(1).getSize() == 20);

        segments.kill();
        segments.append(*LINK(opt0_20));
        segments.append(*LINK(opt20_10));
        segments.append(*LINK(nonOpt30_10));
        CPPUNIT_ASSERT(segments.ordinality() == 3);
        CPPUNIT_ASSERT(!segments.item(0).isWild());
        CPPUNIT_ASSERT(!segments.item(1).isWild());
        CPPUNIT_ASSERT(segments.item(1).getOffset() == 20);
        CPPUNIT_ASSERT(segments.item(1).getSize() == 10);
#endif
    }
};

CPPUNIT_TEST_SUITE_REGISTRATION( SegmentMonitorTest );
CPPUNIT_TEST_SUITE_NAMED_REGISTRATION( SegmentMonitorTest, "SegmentMonitorTest" );

#endif