0.9.9e/pw/dbo.c

/*
        The Malete project - the Z39.2/Z39.50 database framework of OpenIsis.
        Version 0.9.x (patchlevel see file Version)
        Copyright (C) 2001-2003 by Erik Grziwotz, erik@openisis.org

        This library is free software; you can redistribute it and/or
        modify it under the terms of the GNU Lesser General Public
        License as published by the Free Software Foundation; either
        version 2.1 of the License, or (at your option) any later version.

        This library 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 Lesser General Public License for more details.

        You should have received a copy of the GNU Lesser General Public
        License along with this library; if not, write to the Free Software
        Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA

        see README for more information
EOH */

/*
        $Id: dbo.c,v 1.26 2004/11/03 17:39:21 kripke Exp $
        the db object
*/

#include "../pw/pw.h"

#define Q_MAX_EXP 500
#define Q_MAX_DEPTH 50

enum { /* nodes in a query expression */
        OP_NULL, OP_HEAD, /* blank node and head pseudo operator */
        /* operators in increasing precedence; level is op>>1 */
        OP_OR,
        OP_AND=4, OP_NOT, /* AND and NOT/WITHOUT */
        OP_TAGS, /* /(...) tagmask, "kind of" unary */
        OP_SAMEREP = 8, OP_SAMETAG, /* ,(F) ;(G) same field repetition/tag */
        OP_DISTMAX, OP_DISTEX, /* ... and $$$ max/ex distance */
        OP_IDENT, /* last operator, really first simple expression */
        T_QRF = 16, /* #n ref */
        /* term and relation */
        T_EQ, T_PFX, /* term equality and prefix match */
        T_GT, T_GTE, /* relational */
        T_LT, T_LTE, /* relational */
        T_CONT, T_RE /* contains and regexp */
};
static const char *opname[] = {
        "null", "head",
        "+", "",
        "*", "^",
        "/", "",
        ",", ";",
        ".", "$",
        "-"
};
static const char *tname[] = {
        "#",
        "=", "%",
        ">", ">=",
        "<", "<=",
        ":", "~"
};
static const int opdim[] = {
        0, 0,
        0, 0,
        0, 0,
        1, 0, /* / pseudo op */
        2, 1, /* , ; */
        3, 3, /* . $ */
        3 /* IDENT does not actually imply a dimension */
};

typedef union Exp Exp; /* query expression */

typedef struct Bop { /* binary operator */
        int  tag;
        int  dst; /* distance */
        Exp *lft; /* index of left ... */
        Exp *rgt; /* ... and right operand */
} Bop;

typedef struct Tfl { /* tag filter */
        int  tag;
        int  len; /* # tags */
        Exp *lft; /* index of left expression, MUST match pos of Bop.lft */
        unsigned short *lst; /* taglist */
} Tfl;

typedef struct Qrf { /* query ref */
        int  tag; /* == T_QRF */
        Qry *qry;
} Qrf;

typedef Fld Trm;

union Exp { /* query expression */
        int tag;
        Bop bop;
        Tfl tfl;
        Qrf qrf;
        Trm trm;
};

struct Qry { /* parsed query and search result */
        Qry      *nxt; /* chained */
        Dbo      *dbo; /* our database */
        lolo      use; /* req of last use */
        unsigned  qid; /* our id */
        char     *txt; /* full unparsed expression */
        unsigned  sln; /* length of search expression, txt[sln] is '?' or 0 */
        unsigned  cur; /* sor */
        unsigned  len; /* of result set */
        lulu     *res; /* result set, 0 for all */
        lulu      skp; /* min rid to consider (skip to) */
        lulu      brk; /* min rid where set size limit was hit */
        Tfl       srt; /* sort filter */
        Exp      *exp; /* parsed search or filter expression */
        char     *buf; /* holding terms and taglists */
        Exp       ebf[Q_MAX_EXP]; /* exp buffer */
};
/* qry->exp is either 0 or ebf:
        after parsing the search part, it is ebf, possibly the OP_NULL expression.
        Then it is set to 0, if there is no '?' (no filter: record ids only),
        or to ebf for a filter expression (OP_NULL selecting all).
*/

typedef struct Prv { /* dbo private data */
        Dbo dbo; /* public */
        Qry qry; /* the null query */
} Prv;


typedef struct { /* level in tree during parsing */
        Exp *exp;
        int  par; /* number of open parentheses */
} Lev;


/* see doc/Query.txt */
#define EXP_ISSIMPLE(e) (OP_IDENT <= (e)->tag)
static int canmark ( Exp *e )
{
        return EXP_ISSIMPLE(e) ? 1
                : (OP_OR == e->tag /*|| OP_NOT == e->tag*/)
                ? canmark(e->bop.lft) && EXP_ISSIMPLE(e->bop.rgt)
                : OP_TAGS == e->tag ? canmark(e->tfl.lft) : 0;
}

/*
        mode is 'c'reate, 'm'ark or 'f'ilter
        dim the required dimension
        return number of buffers required
*/
static int explain (Exp *e, char mode, int dim, unsigned short tag)
{
        char lmode, rmode;
        int lcost, rcost, cost;
        if ( T_QRF <= e->tag ) {
                if ( T_QRF < e->tag )
                        eRr(LOG_VERBOSE, "%c%d %s \"%.*s\"/%d",
                                mode, dim, tname[e->tag-T_QRF], e->trm.len, e->trm.val, tag);
                else if ( tag || dim ) {
                        eRr(ERR_INVAL, "tag %d dim %d at ref#%d", tag, dim, e->qrf.qry->qid);
                        return -1;
                } else
                        eRr(LOG_VERBOSE, "%c%d #%d", mode, dim, e->qrf.qry->qid);
                return 'm' == mode ? 0 : 1;
        }
        if ( OP_TAGS == e->tag ) {
                eRr(LOG_VERBOSE, "/(%d,..)[%d]", *e->tfl.lst, e->tfl.len);
                return e->tfl.lft ? explain(e->tfl.lft, mode, dim, *e->tfl.lst) : 0;
        }
        if ( OP_DISTMAX == (e->tag&~1) )
                eRr(LOG_VERBOSE, "%c%d %s%d", mode, dim, opname[e->tag], e->bop.dst);
        else
                eRr(LOG_VERBOSE, "%c%d %s", mode, dim, opname[e->tag]);
        if ( !e->tag )
                return 0;
        /* bin op */
        lmode = rmode = mode;
        cost = 0;
        if ( OP_OR == e->tag ) {
                if ( 'f' == mode )
                        cost = 1; /* because we need to keep the filter */
        } else if (canmark(e->bop.rgt)) {
                rmode = 'm';
        } else {
                lmode = 'f';
                if ( 'f' == mode )
                        cost = 1; /* double filtering */
        }
        if ( dim < opdim[e->tag] )
                dim = opdim[e->tag];
        lcost = e->bop.lft ? explain(e->bop.lft, lmode, dim, tag) : 0;
        if ( 0 > lcost ) return lcost;
        rcost = e->bop.rgt ? explain(e->bop.rgt, rmode, dim, tag) : 0;
        if ( 0 > rcost ) return rcost;
        cost += lcost > rcost ? lcost : rcost;
        return cost;
}       /* explain */

/*
        parse search or filter expression
        tokens are appended to an array of expressions.
        Operators form a tree in this array, with 0 being the head.
        We always operate on right childs, which, however,
        may become the left child of following operators.
        We maintain a path of right childs up to the current node.
        The current node can either be blank or a closed expression
        (after a term or closing parenthesis).
        If the next token is
        - an operator, the current node must not be blank for most operators
                (although some may use a default expression as left hand).
                While nodes on the path have same or higher precedence than the operator,
                and no node with open parentheses is encountered,
                the path is unwound. The then current node becomes the
                operators left child and is replaced by the operator.
                The operator opens a blank node as right child.
        -       a terminal (term or ref),
                if the current node is not blank, a default AND is inserted.
                Then we have a blank node, which becomes the terminal.
        - a left parenthesis also produces a default AND, if the current
                node is not blank. Then the blank nodes parentheses count is increased.
        - a right parenthesis behaves like an operator in that the
                current node must not be blank. It then unwinds the path up to
                the first node whose parentheses count is not zero and decreases it.
        buffer usage: query length is sufficient.
        - quoted terms, without the quotes, which is 2 bytes less than
                in the query expression. may run on the 0 byte in unterminated literal,
                however, the inital quote accounts for that.
        - 2 bytes per tag in tagfilter. since every tag is at least 1 digit
                plus 1 extra character (the / or terminating ,), needs not more bytes
                then in the query expression. we may have to align to 2 after a
                proper quoted term, but the terms closing quote accounts for that.
        This may be called twice:
        - on the full query text, with q->exp unset and q->txt == txt
        - after the ?, reusing q->exp
*/
static int parse (Qry *q, unsigned char *txt)
{
        Lev path[Q_MAX_DEPTH], *cur = path;
        Exp *exp;
        int n/*# used nodes*/, cost;
        unsigned char *buf = (unsigned char *)q->buf
                + q->sln; /* buffer might still be used by q->srt */
        unsigned char *p = txt, *pos=p;

        eRr(LOG_VERBOSE, "parsing '%s' as %s",
                txt, q->exp ? "filter" : "search");
        if ( q->exp ) /* in filtering -- reuse exp buffer */
                memset(exp = q->exp, 0, Q_MAX_EXP * sizeof(Exp));
        else
                exp = q->ebf;
        memset(path, 0, sizeof(path));
        /* add the pseudo op head with lowest precedence */
        cur->exp = exp;
        exp->tag = OP_HEAD;
        exp->bop.rgt = (++cur)->exp = exp+1; /* add blank */
        n = 2;
#define QRAP( args ) do { eRr args; goto qrap; } while (0) /* query is crap */
        for (;;) {
                Exp tok;
                /* get a token */
                while (33 > *p) if ( !*p++ ) { p--; goto done; } /* skip white */
                pos = p;
                LOG_DBG(LOG_DEBUG, "pos %d lev %d: %c", pos-txt, cur-path, *p);
                switch (*p) {
                case '+': tok.tag = OP_OR; p++; break;
                case '*': tok.tag = OP_AND; p++; break;
                case '^': tok.tag = OP_NOT; p++; break;
                case ';': tok.tag = OP_SAMETAG; p++; break;
                case ',': tok.tag = OP_SAMEREP; p++; break;
                case '.': tok.tag = OP_DISTMAX;
                        for ( tok.bop.dst=1; '.' == *++p; ) tok.bop.dst++;
                        break;
                case '$': tok.tag = OP_DISTEX;
                        for ( tok.bop.dst=1; '$' == *++p; ) tok.bop.dst++;
                        break;
                case '-': tok.tag = OP_IDENT; p++; break;
                case ')':
                        eRr(LOG_VERBOSE, ") on lev %d", cur-path);
                        if ( !cur->exp->tag )
                                QRAP((ERR_INVAL,"unexpected )"));
                        while ( !cur->par )
                                if ( path == --cur )
                                        QRAP((ERR_INVAL,"unbalanced )"));
                        cur->par--;
                        eRr(LOG_VERBOSE, "now on lev %d", cur-path);
                        p++;
                        continue;
                case '/': {
                        int multi = '(' == *++p;
                        unsigned short *lst;
                        if ( multi )
                                p++;
                        tok.tag = OP_TAGS;
                        if ( 1 & (int)buf ) buf++; /* align */
                        lst = tok.tfl.lst = (unsigned short*)buf;
                        for (;;) {
                                int tag, got;
                                while (*p && 33 > *p) p++; /* skip white */
                                got = a2il((char*)p, -1, &tag);
                                if ( !got )
                                        QRAP((ERR_INVAL, "/ expects digit"));
                                if ( tag<0 || 65535<tag )
                                        QRAP((ERR_INVAL, "tag %d out of range 0..65535", tag));
                                *lst++ = (unsigned short)tag;
                                p += got;
                                if ( !multi )
                                        break;
                                while (*p && 33 > *p) p++; /* skip white */
                                if ( ')' == *p++ )
                                        break;
                                if ( ',' != p[-1] )
                                        QRAP((ERR_INVAL, "/ expects ,"));
                        }
                        tok.tfl.len = lst - tok.tfl.lst;
                        buf = (unsigned char*)lst;
                        if ( !cur->exp->tag && 2==n ) { /* save special case */
                                tok.tfl.lft = 0;
                                if ( q->exp ) /* parsing filter */
                                        *(exp->bop.lft = exp + n++) = tok; /* as left of head */
                                else
                                        q->srt = tok.tfl; /* as sort expression */
                                continue;
                        }
                        break;
                        } /* tags */
                case '?':
                        if ( q->exp )
                                QRAP((ERR_INVAL, "unexpected ? in filter"));
                        goto done;
                case '(':
                        if ( ('F' == p[1] || 'G' == p[1]) && ')' == p[2] ) { /* legacy ops */
                                tok.tag = 'F' == p[1] ? OP_SAMEREP : OP_SAMETAG;
                                p += 3;
                                break;
                        }
                        if ( CT_D == lat1ct[p[1]] ) { /* (n) */
                                int dst, dig = a2il((char*)p+1,-1,&dst);
                                if ( ')' == p[1+dig] ) {
                                        tok.tag = OP_DISTMAX;
                                        tok.bop.dst = dst;
                                        p += dig+2;
                                        break;
                                }
                        }
                default: /* terminal or parenthesis */
                        if ( cur->exp->tag ) { /* add default AND */
                                tok.tag = OP_AND;
                                break;
                        }
                        if ( '(' == *p ) {
                                eRr(LOG_VERBOSE, "( on lev %d", cur-path);
                                cur->par++;
                                p++;
                                continue;
                        }
#ifdef REFS_ARE_NOW_FIXED
                        if ( '#' == *p ) {
                                unsigned qid;
                                Qry *qq;
                                if ( CT_D != lat1ct[*++p] )
                                        QRAP((ERR_IDIOT,"#expects digit"));
                                p += a2il((char*)p, -1, (int*)&qid);
                                for (qq = q->dbo->qry; qq && qid != qq->qid; qq = qq->nxt)
                                        ;
                                if ( !qq )
                                        QRAP((ERR_BADF, "query #%u not found (maybe expired)", qid));
                                if ( q->exp )
                                        QRAP((ERR_IDIOT, "sorry, refs supported in search only"));
                                tok.tag = T_QRF;
                                tok.qrf.qry = qq;
                                *cur->exp = tok;
                                continue;
                        }
#endif
                        /* check for relation */
                        tok.tag = T_EQ;
                        switch (*p) {
                        case '=': p++; break;
                        case '%': tok.tag = T_PFX; p++; break;
                        case '>':
                                if ( '=' != *++p ) tok.tag = T_GT; else { tok.tag = T_GTE; p++; }
                                break;
                        case '<':
                                if ( '=' != *++p ) tok.tag = T_LT; else { tok.tag = T_LTE; p++; }
                                break;
                        case ':':
                                if ( !q->exp )
                                        QRAP((ERR_INVAL, ": only in filter"));
                                tok.tag = T_CONT; p++;
                                break;
                        case '~':
                                /* tok.tag = T_RE; p++; break; */
                                QRAP((ERR_INVAL, "~ unsupported"));
                        }
                        while (33 > *p) if ( !*p++ ) goto done; /* skip white */
                        if ( '"' != *p ) { /* plain - point to expr */
                                if ( CT_S == lat1ct[*p] )
                                        QRAP((ERR_INVAL,"bad character '%c'",*p));
                                tok.trm.val = (char*)p;
                                while (!((CT_WHITE|CT_SPECL) & lat1ct[*p++]))
                                        ;
                                tok.trm.len = (char*)--p - tok.trm.val;
                        } else { /* quoted - use buffer */
                                tok.trm.val = (char*)buf--;
                                while ((*++buf = *++p) && ('"'!=*buf || '"'==*++p))
                                        ;
                                if ( !*buf )
                                        QRAP((ERR_INVAL,"unexpected end of query in literal"));
                                tok.trm.len = (char*)buf - tok.trm.val;
                        }
                        if ( '$' == *p ) {
                                tok.tag = T_PFX;
                                p++;
                        }
                        *cur->exp = tok;
                        continue;
                }
                /* not continued: got operator */
                eRr(LOG_VERBOSE, "%s on lev %d", opname[tok.tag], cur-path);
                if ( !cur->exp->tag )
                        QRAP((ERR_INVAL,"unexpected operator '%s'", opname[tok.tag]));
                if ( OP_DISTMAX > tok.tag )
                /* the DISTs are right associative with highest precedence */
                /* for all others, unwind path up to lower precedence or parenthesis */
                while ( !cur->par )
                        if ( ((--cur)->exp->tag&~1) < (tok.tag&~1) ) {
                                cur++;
                                break;
                        }
                /* now cur->exp is the victim to become our left child */
                tok.bop.lft = cur->exp;
                cur->exp = cur[-1].exp->bop.rgt = exp + n++; /* get new op exp */
                if ( n > Q_MAX_EXP-1 ) /* allow one for right child */
                        QRAP((ERR_INVAL, "max #expressions %d exceeded", Q_MAX_EXP));
                *cur->exp = tok;
                if ( OP_TAGS != tok.tag ) {
                        Exp *child = cur->exp->bop.rgt = exp + n++; /* empty right child */
                        if ( (++cur - path) == Q_MAX_DEPTH )
                                QRAP((ERR_INVAL, "max depth %d exceeded", Q_MAX_DEPTH));
                        cur->exp = child;
                        cur->par = 0;
                }
        }       /* for ever */
done:
        if ( !cur->exp->tag && n>(exp->bop.lft ? 3 : 2) )
                /* ok only if completely empty */
                QRAP((ERR_INVAL,"unexpected end of query after operator"));
        /* to be strict, we should check for open parenthesis -- but who cares */
        if ( !exp->bop.lft ) /* purge head unless we have a tagfilter */
                *exp = *exp->bop.rgt;
        cost = explain(exp,'c',0,0);
        if ( 0 > cost )
                QRAP((ERR_INVAL,"no plan"));
        eRr(LOG_VERBOSE, "successfully parsed %d bytes cost %d", p - txt, cost);
        q->exp = exp;
        return p - txt;
qrap: /* cleanup */
#undef QRAP
        eRr(ERR_INVAL, "error in query '%s' near pos %d", txt, pos-txt);
        q->exp = 0;
        return ERR_INVAL;
}       /* parse */


typedef struct { /* long long pointer */
        lulu     rit; /* higher 6 rid, lower two tag */
        unsigned pos;
} Llp;
#define RIDMSK LULU(0xffffffffffff0000)
#define OCCMSK 0xffff0000U

#define PGT(a, b) \
        ((a)->rit > (b)->rit ? 1 : (a)->rit < (b)->rit ? 0 : (a)->pos > (b)->pos)


typedef struct Res {
        int   len;
        Llp   ptr[QDX_MAXVALPERLEAF]; /* really ses->s */
} Res; /* followed by ses->s marks */
#define RES_MARKS(res) (((char*)(res))+4+ses->s*sizeof(Ptr))


/* TODO: maintain pool */
static Res *rAlloc ()
{
        return mAlloc(4 + ses->s*(sizeof(Ptr)+1));
}


static void rFree (Res *res)
{
        mFree(res);
}


typedef struct { /* a filter */
        Res  *res; /* results against which to filter */
        Bop  *cnd; /* matching condition */
        char *mrk; /* optional mark array */
        char  set; /* the mark to set/ positive or negative filtering */
} Flt;


/*
        match res against flt on cnd
        with mrk, set mrk to set for corresponding entries in flt
        else compact res, keeping matches is set, nonmatches else
        TODO: prolly we'd rather use a copy of flt; however, compiler might do this
*/
static void rMatch (Res *res, Flt *flt)
{
        Llp *p = res->ptr, *pe = p + res->len, *d = p;
        Llp *f = flt->res->ptr, *fe = f + flt->res->len;
        unsigned dim = opdim[flt->cnd->tag], pos; /* filter dim, pos */
        char *m;
#ifndef NDEBUG
        int mcnt = 0;
#       define MCNT mcnt++;
#else
# define MCNT
#endif

        eRr(LOG_VERBOSE,
                "matching res len %d against flt len %d cnd %d set %d mrk %d dim %d",
                res->len, flt->res->len, flt->cnd->tag, flt->set, 0!=flt->mrk, dim);
        if ( !flt->res->len )
                goto done;
        /*
                filter by sweep, costing O(flt->res->len + res->len)
                for res->len << flt->res->len/ld(flt->res->len), binsearch could be cheaper
                however, sweep is much simpler with varying dimensions
                and for marking, we'll do a full pass anyway
                TODO: lotsa other optimizations possible here
                obviously we could pull several case distinctions out of the loop
                to trade code size for speed
        */

        for ( ; p < pe; p++ ) {
                /* advance f, while it is too low */
                lulu minrit = p->rit;
                if ( !dim ) minrit &= RIDMSK;
                while ( f->rit < minrit ) if ( fe == ++f ) goto done;
                if ( (RIDMSK&f->rit) > p->rit ) goto nomatch;
                if ( !dim ) goto match;
                if ( f->rit > p->rit ) goto nomatch;
                if ( 1 == dim ) goto match;
                /* got same rit: rid+tag */
                pos = OCCMSK & p->pos;
                if ( 3 == dim && pos < p->pos - flt->cnd->dst )
                        pos = p->pos - flt->cnd->dst;
                while ( f->pos < pos ) {
                        if ( fe == ++f ) goto done;
                        if ( f->rit != p->rit ) goto nomatch;
                }
                /* still same rit */
                if ( f->pos > (p->pos|0xffff) ) goto nomatch; /* occ mismatch */
                /* and same occ */
                if ( 2 == dim ) goto match;
                if ( f->pos > p->pos + flt->cnd->dst ) goto nomatch; /* far out */
                if ( OP_DISTMAX == flt->cnd->tag ) goto match;
                /* the strange OP_DISTEX -- alas */
                if ( f->pos == p->pos - flt->cnd->dst ) { /* lower match */
                        if ( !flt->mrk ) goto match;
                        flt->mrk[f - flt->res->ptr] = flt->set;
                }
                if ( !(OCCMSK & (p->pos ^ (pos = p->pos + flt->cnd->dst))) ) { /* no ovfl */
                        Llp *f1 = f; /* scan for upper match */
                        while ( pos != f1->pos ) {
                                if ( fe == ++f1 || f1->rit != p->rit || f1->pos > pos ) goto nomatch;
                        }
                        if ( !flt->mrk ) goto match;
                        flt->mrk[f1 - flt->res->ptr] = flt->set;
                }
                /* had nomatch or marked marking match: go on */
        nomatch:
                if ( flt->mrk ) continue;
                if ( flt->set ) continue;
                goto addit;
        match:  /* filter matched */
                if ( !flt->mrk ) {
                        if ( flt->set ) {
        addit:
                                if ( d < p ) *d = *p;
                                d++;
                        }
                        continue;
                }
                LOG_DBG(LOG_DEBUG, "marking");
                /* marking */
                m = flt->mrk + (f - flt->res->ptr);
                switch (dim) {
                case 0:
                        while ( !(RIDMSK & (p->rit ^ f->rit)) ) {
                                MCNT *m++ = flt->set; if ( fe == ++f ) goto done;
                        }
                        break;
                case 1:
                        while ( p->rit == f->rit ) {
                                MCNT *m++ = flt->set; if ( fe == ++f ) goto done;
                        }
                        break;
                case 2:
                        while ( p->rit == f->rit && !(OCCMSK & (p->pos^f->pos)) ) {
                                MCNT *m++ = flt->set; if ( fe == ++f ) goto done;
                        }
                        break;
                case 3: /* DISTMAX only -- DISTEX done above */
                        if ( OCCMSK & (p->pos ^ (pos = p->pos + flt->cnd->dst)) ) /* overflow */
                                pos = p->pos | 0xffff;
                        while ( p->rit == f->rit && !(OCCMSK & (p->pos^f->pos)) ) {
                                MCNT *m++ = flt->set; if ( fe == ++f ) goto done;
                        }
                        break;
                }
        }       /* for p < pe */
done:
        if ( flt->mrk ) {
#ifndef NDEBUG
                eRr(LOG_VERBOSE, "marked %d", mcnt);
#endif
                return;
        }
        if ( !flt->set && p < pe ) { /* negative filter exhausted */
                if ( d == p ) return; /* none filtered -- keep full set */
                memmove(d, p, (pe-p)*sizeof(*p));
                d += pe-p;
        }
        res->len = d - res->ptr;
}       /* rMatch */


/*
        projection to rit (2 dimensions rid+tag)
*/
static void rProj2 (Res *res)
{
        Llp *p = res->ptr, *e = p + res->len, *d;
        lulu last = 0;

        for (;;) {
                if ( e == p ) return;
                if ( last == p->rit ) break;
                last = p++->rit;
        }
        /* now p is 1st redundant */
        for (d = p; ++p < e;)
                if ( last != p->rit ) {
                        last = p->rit;
                        *d++ = *p++;
                }
        res->len = d - res->ptr - 1;
}       /* rProj2 */


static lulu rMerge (Res *res, Res *add, int dim)
{
        lulu brk = 0;
        Llp *a = res->ptr + res->len-1, *end = res->ptr + ses->s;
        Llp *b, *set;
        if ( !res->len ) {
                if ( 2 > dim ) rProj2(res);
                memcpy(res, add, 4+add->len*sizeof(Ptr));
                return res->len;
        }
        if ( 3 != dim ) { /* reduce the add set */
                static const int tag[] = {OP_AND, OP_SAMETAG, OP_SAMEREP};
                Bop bop;
                Flt flt;
                bop.tag = tag[dim];
                flt.res = res;
                flt.cnd = &bop;
                flt.mrk = 0;
                flt.set = 0;
                rMatch(add, &flt);
        }
        LOG_DBG(LOG_DEBUG, "merging %d+%d dim %d", res->len, add->len, dim);
        b = add->ptr + add->len-1;
        for (set = a + add->len; set > a; set--) {
                Llp *gt = PGT(a, b) ? a : b;
#if 0
                LOG_DBG(LOG_DEBUG, "a %d b %d set %d use %c",
                        a+1-res->ptr, b+1-add->ptr, set-res->ptr, b==gt ? 'b' : 'a');
#endif
                if ( set < end )
                        *set = *gt;
                else
                        brk = gt->rit;
                if ( b == gt )
                        b--; /* if used last b, should bail out on set == a */
                else if (res->ptr == a--) { /* used last a */
                        /* now the remaining bs are to become first in result */
                        LOG_DBG(LOG_DEBUG, "copying %d bs to %d slots",
                                b+1-add->ptr, set-res->ptr);
                        memcpy(res->ptr, add->ptr, ((char*)(b+1)) - ((char*)add->ptr));
                        break;
                }
        }
        LOG_DBG(LOG_DEBUG, "merged %d+%d brk %d untouched %d b %d",
                res->len, add->len, (unsigned)(brk>>16), set+1-res->ptr, b+1-add->ptr);
        if ( !brk ) {
                res->len += add->len;
                return 0;
        }
        /* kill entries on brk rid -- find last valid */
        brk &= ~RIDMSK;
        for ( end = res->ptr+ses->s; end-- > res->ptr && brk < end->rit; ) ;
        res->len = end + 1 - res->ptr;
        return brk;
}       /* rMerge */


typedef struct { /* search context */
        QLoop qlp;
        Res  *res; /* array to fill in callback */
        Qry  *qry; /* query */
        Tfl  *tfl; /* current tag filter */
        int   dim; /* required dimension */
        Flt   flt; /* filter or mark set */
        int   dsc; /* discard filter */
} Scx;


static int scb (Scx *scx) /* qLoop callback: search one term */
{
        Res tmp; /* merge buffer */
        unsigned i;
        const unsigned char *v = scx->qlp.vals;
        int typ = scx->qlp.qdx->ptr;
        Llp *p = tmp.ptr;

        eRr(LOG_VERBOSE, "scb '%b' %d", scx->qlp.cur.len, scx->qlp.cur.byt,
                scx->qlp.nvals);
        /* for tag-sorted (non FULTXT) index, we should pre-skip to tag */
        for ( i=scx->qlp.nvals; i--; ) {
#ifdef CPU_BIG_ENDIAN
enum { RID5, RID4, RID3, RID2, RID1, RID0, TAG1, TAG0 };
enum { POS3, POS2, POS1, POS0 };
#else
enum { TAG0, TAG1, RID0, RID1, RID2, RID3, RID4, RID5 };
enum { POS0, POS1, POS2, POS3 };
#endif
                unsigned char *rit = (unsigned char*)&p->rit;
                unsigned char *pos = (unsigned char*)&p->pos;
                p->rit = p->pos = 0;
                if ( !(QDX_FULTXT & typ) ) /* leading tag */
                        switch ( QDX_TAGMSK&typ ) {
                        case QDX_TAG2: rit[TAG1] = *v++;
                        case QDX_TAG1: rit[TAG0] = *v++;
                        }
                switch ( QDX_RIDMSK&typ ) {
                case QDX_RID6: rit[RID5] = *v++;
                case QDX_RID5: rit[RID4] = *v++;
                case QDX_RID4: rit[RID3] = *v++;
                case QDX_RID3:
                        rit[RID2] = *v++;
                        rit[RID1] = *v++;
                        rit[RID0] = *v++;
                }
                if ( QDX_FULTXT & typ )
                        switch ( QDX_TAGMSK&typ ) {
                        case QDX_TAG2: rit[TAG1] = *v++;
                        case QDX_TAG1: rit[TAG0] = *v++;
                        }
                switch ( QDX_POSMSK&typ ) {
                case 4: pos[POS3] = *v++;
                case 3: pos[POS2] = *v++;
                case 2: pos[POS1] = *v++;
                case 1: pos[POS0] = *v++;
                }
                /*
                eRr(LOG_VERBOSE, "rid %d tag %d pos %d",
                        (int)(p->rit >> 16), (int)(0xffff & p->rit), p->pos);
                */
                if ( !scx->tfl ) goto tagok;
                /* find in tag filter */
                if ( 1 == scx->tfl->len ) {
                        if ( (unsigned short)p->rit == *scx->tfl->lst ) goto tagok;
                } else {
                        unsigned short *t = scx->tfl->lst, *te = t + scx->tfl->len;
                        while ( t < te ) if ( (unsigned short)p->rit == *t++ ) goto tagok;
                }
                continue;
        tagok:
                if ( p->rit < scx->qry->skp ) continue;
                if ( scx->qry->brk && p->rit >= scx->qry->brk ) break;
                p++;
        }
        if ( !(tmp.len = p - tmp.ptr) ) goto done;
        if ( scx->flt.res ) {
                rMatch(&tmp, &scx->flt);
                if ( scx->flt.mrk ) goto done;
        }
        if ( scx->res->len )
                rMerge(scx->res, &tmp, scx->dim);
        else
                memcpy(scx->res, &tmp, 4+tmp.len*sizeof(Llp));
done:
        return 0;
}       /* scb */


static Res *search (Scx *scx, Exp *exp) /* search one expression */
{
        Res *a = 0, *b;
        int dim;

        if ( scx->flt.res )
                eRr(LOG_VERBOSE, "%sing %d %c on %d", scx->flt.mrk?"mark":"filter",
                        exp->tag, scx->flt.set?'+':'-', scx->flt.cnd->tag);
        else
                eRr(LOG_VERBOSE, "creating %d", exp->tag);
        if ( OP_IDENT <= exp->tag ) { /* simple expression */
                scx->qlp.flg = 0;
                if ( OP_IDENT < exp->tag ) {
                        qMkKey(scx->qlp.qdx, &scx->qlp.key, exp->trm.val, exp->trm.len);
                        switch (exp->tag) {
                        case T_EQ: scx->qlp.flg = QEQ; break;
                        case T_PFX: scx->qlp.flg = QPF; break;
                        }
                } else {
                        scx->qlp.flg = QUPTO;
                        qMkKey(scx->qlp.qdx, &scx->qlp.key,
                                exp->bop.lft->trm.val, exp->bop.lft->trm.len);
                        qMkKey(scx->qlp.qdx, &scx->qlp.to,
                                exp->bop.rgt->trm.val, exp->bop.rgt->trm.len);
                }
                if ( !(QDX_FULTXT & scx->qlp.qdx->ptr) ) {
                        if (!scx->tfl || 1 != scx->tfl->len) {
                                eRr(ERR_INVAL, "need tag filter with non-fulltext db");
                                return 0;
                        }
                }
                if ( !scx->flt.mrk )
                        scx->res = a = rAlloc();
                qLoop(&scx->qlp);
                if ( a )
                        eRr(LOG_VERBOSE, "got %d", a->len);
                if ( scx->dsc ) {
                        rFree(scx->flt.res);
                        scx->flt.res = 0;
                }
                return a;
        }
        switch ( exp->tag ) {
        case OP_NULL:
                return 0;
        case OP_HEAD:
                return search(scx, exp->bop.rgt);
        case OP_OR: {
                int dsc = scx->dsc;
                scx->dsc = 0; /* delay discarding filter */
                a = search(scx, exp->bop.lft);
                scx->dsc = dsc;
                b = search(scx, exp->bop.rgt);
                if ( b ) {
                        if ( a )
                                rMerge(a, b, scx->dim);
                        rFree(b);
                }
                eRr(LOG_VERBOSE, "got %d", a->len);
                return a;
        } /* case OR */
        case OP_TAGS: {
                Tfl *tfl = scx->tfl;
                scx->tfl = &exp->tfl;
                a = search(scx, exp->tfl.lft);
                scx->tfl = tfl;
                return a;
        } /* case TAGS */
        /*
        case OP_NOT: TODO: special case marking not impl.
                return 0;
        */
        default:
                if ( (dim = scx->dim) < opdim[exp->tag] )
                        scx->dim = opdim[exp->tag];
                /* if not filtering and the right hand does not support direct marking,
                        create it and use as left hand filter */
                if ( !scx->flt.res && !canmark(exp->bop.rgt) ) {
                        scx->flt.res = b = search(scx, exp->bop.rgt); /* create right hand */
                        scx->flt.cnd = &exp->bop;
                        scx->flt.mrk = 0;
                        scx->flt.set = OP_NOT != exp->tag;
                        a = search(scx, exp->bop.lft);
                        scx->flt.res = 0;
                        break;
                } /* Else create the left hand (using any filter) */
                if ( (a = search(scx, exp->bop.lft)) ) {
                        Flt sav, use;
                        use.res = a;
                        use.cnd = &exp->bop;
                        use.mrk = RES_MARKS(a);
                        use.set = OP_NOT != exp->tag;
                        memset(use.mrk, !use.set, ses->s);
                        sav = scx->flt;
                        /* evaluate the right hand */
                        if ( canmark(exp->bop.rgt) ) { /* using direct marking, if possible */
                                scx->flt = use;
                                search(scx, exp->bop.rgt); /* direct marking on a */
                        } else { /* create it */
                                scx->flt.res = 0; /* non filtering */
                                if ( (b = search(scx, exp->bop.rgt)) ) {
                                        rMatch(b, &use); /* and mark */
                                        rFree(b);
                                }
                        }
                        if ( a->len ) { /* reduce the marked set */
                                Llp *p = a->ptr, *e = p + a->len, *d;
                                char *m = use.mrk;
                                eRr(LOG_VERBOSE, "reducing a->len %d", a->len);
                                while ( *m++ ) if ( e == ++p ) goto done;
                                eRr(LOG_VERBOSE, "1st %d were good", p - a->ptr);
                                /* now *p is 1st unmarked */
                                for ( m--, d=p; p<e; p++ ) if ( *m++ ) *d++ = *p;
                                a->len = d - a->ptr;
                                eRr(LOG_VERBOSE, "... to a->len %d", a->len);
                        }
                done:
                        scx->flt = sav;
                }
                ;
        }
        scx->dim = dim;
        return a;
}       /* search */


/*
        parse query and execute search from 'Q<TAB>*' msg
*/
static Qry *Q ( Dbo *self, Msg msg )
{
        Qry *q = 0, *p;
        char *exp = msg->val+2;
        int l = msg->len-2, i = 0;

        if ( !ses->qry )
                ses->qry = mAlloc(ses->q * sizeof(Qry));
        if ( !self->qry )
                self->qry = &((Prv*)self)->qry;
#ifdef REFS_ARE_NOW_FIXED
        if ( l && '#' == *exp && l-1 == a2il(exp+1,l-1,&i) ) { /* plain #n ref */
                p = self->qry;
                if ( 1 == l )
                        ((Prv*)self)->qry.cur = 0; /* reset 0 query */
                if ( (unsigned)i == p->qid )
                        return p;
                for ( ; (q = p->nxt); p=q )
                        if ( (unsigned)i == q->qid ) {
                                p->nxt = q->nxt; /* unlink */
                                q->nxt = self->qry; /* put at front */
                                return self->qry = q;
                        }
                return 0;
        }
#endif
        /* create new query */
        p=q=ses->qry + ses->q -1; do { /* find slot */
                if ( !p->dbo ) { /* unused */
                        q = p;
                        goto gotslot;
                }
                if ( q->use < p->use )
                        q = p;
        } while ( ses->qry < --p );
        /* unlink the lrused */ {
                Dbo *dbo = q->dbo;
                if ( q == dbo->qry )
                        dbo->qry = 0; /* kill current */
                else {
                        for (p = dbo->qry; q != p->nxt; )
                                if ( !(p = p->nxt) ) {
                                        eRr(ERR_IDIOT, "unlinked");
                                        return 0;
                                }
                        p->nxt = q->nxt; /* unlink */
                }
                mFree(q->txt);
                mFree(q->res);
                mFree(q->buf);
        }
        memset(q, 0, sizeof(*q));
gotslot: /* got clean q */
        q->dbo = self;
        q->qid = self->qid+1;
        q->buf = mAlloc(l); /* see notes on buffer usage at parse */
        if ( 0 > (i = parse(q, (unsigned char*)(q->txt = mDupz(exp, l)))) )
                goto cleanup;
        q->sln = i;
        q->skp = 0x10000; /* always skip rid 0 entries */
        if ( !q->exp->tag ) /* the NULL search: loop all */
                q->len = self->db->rdx.mid;
        else {
                Scx scx;
                Res *res;
                lulu last = 0, *d;
                Llp *r, *e;

                memset(&scx, 0, sizeof(scx));
                scx.qlp.qcb = (QCb*)scb;
                scx.qlp.qdx = &self->db->qdx;
                scx.qry = q;
                if ( !(res = search(&scx, q->exp)) || !res->len )
                        goto cleanup;
                d = q->res = mAlloc(res->len * sizeof(lulu));
                r = res->ptr;
                e = r + res->len;
                do
                        if ( last != (*d = r->rit >> 16) ) last = *d++;
                while (++r < e);
                q->len = d - q->res;
        }
        if ( '?' == q->txt[i] )
                i = parse(q, (unsigned char*)q->txt+i+1);
        else {
                q->exp = 0;
                MFREE(q->buf);
        }
        self->qid++;
        q->nxt = self->qry; /* link */
        self->qry = q;
        return q;
cleanup:
        mFree(q->txt);
        mFree(q->res);
        mFree(q->buf);
        memset(q, 0, sizeof(*q));
        return 0;
}       /* Q */


/*
        apply filter expression to record rid.
        return 1, if record passed, else 0.
*/
static int filter ( Dbo *self, lulu rid, Exp *e )
{
        if ( !e ) {
                eOut(0, "%u", (unsigned)rid);
                return 1;
        }
        if ( !e->tag ) {
                if ( 0 <= rRead(&env.out->lst, &self->db->rdx, rid, self->p) ) {
                        SINK(env.out);
                        return 1;
                }
                return 0;
        }
        return 1;
}       /* filter */


typedef struct { /* search context */
        QLoop qlp;
        int   lim; /* limit #terms */
        int   cnt; /* count for term so far */
        int   det; /* detailled: restrict to tag, count precise rids */
        int   rln; /* length of rid: 3..6 */
        int   tln; /* length of tag to match: 0..2 */
        int   rof; /* offset of rid in vals: 0 or taglength */
        int   tof; /* offset of tag in vals: 0 or ridlength */
        char  rid[6]; /* last rid seen */
        char  tag[2]; /* tag to filter on */
        char  key[512]; /* decoded term */
        int   kln; /* key length */
} Tcx;


static int tcb (Tcx *tcx) /* qLoop callback: list terms */
{
        const unsigned char *v, *e;
        char *rid;
        int vsz;

        if ( !(QSAME & tcx->qlp.flg) ) { /* new term */
                if ( tcx->cnt )
                        eOut(0, "%d\011%.*s", tcx->cnt, tcx->kln, tcx->key);
                if ( !tcx->lim-- )
                        return 1; /* break loop */
                tcx->kln = qRdKey(tcx->qlp.qdx, tcx->key, sizeof tcx->key, &tcx->qlp.cur);
                tcx->cnt = 0;
        }
        if ( !tcx->det ) {
                tcx->cnt += tcx->qlp.nvals;
                return 0;
        }
        /* detailled */
        rid = tcx->rid;
        if ( !(QSAME & tcx->qlp.flg) )
                memset(rid, 0, tcx->rln);
        vsz = tcx->qlp.qdx->vsz;
        v = tcx->qlp.vals;
        e = v + tcx->qlp.nvals*vsz;
        for ( ; v < e; v += vsz )
                if ( !(tcx->tln && memcmp(v+tcx->tof, tcx->tag, tcx->tln))
                        && memcmp(v+tcx->rof, rid, tcx->rln)
                ) {
                        tcx->cnt++;
                        rid = (char*)v+tcx->rof;
                }
        if ( tcx->rid != rid )
                memcpy(tcx->rid, rid, tcx->rln);
        return 0;
}       /* tcb */


/*
*/
static int T ( Db *db, Fld *msg ) /* terms */
{
        char *v = msg->val+2, *e = msg->val + msg->len, *p;
        Tcx tcx;

        memset(&tcx, 0, sizeof(tcx));
        tcx.qlp.qcb = (QCb*)tcb;
        tcx.qlp.qdx = &db->qdx;
        tcx.qlp.flg = QPF;
        if ( !(tcx.lim = ses->s) )
                tcx.lim = 1;
        if ( v < e ) {
                for (p = v; v < e && TAB != *v; v++) ;
                qMkKey(tcx.qlp.qdx, &tcx.qlp.key, p, v-p);
                if ( v++ < e ) {
                        for (p = v; v < e && TAB != *v; v++) ;
                        if ( v == p )
                                tcx.qlp.flg = QLOOP; /* endless */
                        else {
                                tcx.qlp.flg = QUPTO;
                                qMkKey(tcx.qlp.qdx, &tcx.qlp.to, p, v-p);
                        }
                        if ( v++ < e ) { /* detailled */
                                int typ = tcx.qlp.qdx->ptr, tag;
                                if ( QDX_LEAFPV & db->qdx.typ ) {
                                        eOut(0, "\011-1\011plain values");
                                        return ERR_INVAL;
                                }
                                a2il(v, e-v, &tag);
                                tcx.det = !0;
                                tcx.rln = QDX_RIDMIN+((QDX_RIDMSK&typ)>>QDX_RIDSH);
                                tcx.tln = (QDX_TAGMSK&typ)>>QDX_TAGSH;
                                if ( !(QDX_FULTXT & typ) )
                                        tcx.rof = tcx.tln;
                                else if ( tag )
                                        tcx.tof = tcx.rln;
                                else
                                        tcx.tln = 0;
                                if ( tcx.tln ) { /* get the tag bytes */
                                        Val val;
                                        Ptr ptr;
                                        ptr.tag = (unsigned short)tag;
                                        qMkVal(&val, &ptr, typ);
                                        memcpy(tcx.tag, val.byt+tcx.tof, tcx.tln);
                                }
                        }
                }
        }
        eOut(0, "#\0110"); /* plain ok comment */
        qLoop(&tcx.qlp);
        if ( tcx.cnt )
                eOut(0, "%d\011%.*s", tcx.cnt, tcx.kln, tcx.key);
        return 0;
}       /* T */


enum {
        IDX_FIELD,
        IDX_WORD,
        IDX_SPLIT
};
typedef struct {
        QSet qst;
        int  mod;
        int  mhd;
} Idx;

/*
        process control fields for X method
        if a tag is found, return it's pos
*/
static char *idxctl (Idx *i, Fld *f)
{
        char *p = f->val, *e = p+f->len, *t = p;
        for ( ; p<e; p = ++t) {
                while (t < e && TAB != *t) t++;
                switch (*p) {
                case 'f': i->mod = IDX_FIELD; break;
                case 'w': i->mod = IDX_WORD;  break;
                case 's': i->mod = IDX_SPLIT; break;
                case 'a': i->qst.del = 0; break;
                case 'd': i->qst.del = 1; break;
                case 'm': i->mhd = t>p+1 && 'H'==p[1]; continue;
                case 'p':
                        if ( t == p+1 )
                                i->qst.pfx.len = 0;
                        else {
                                int len = t-p-1;
                                if ( 255 < len ) len = 255;
                                memcpy(i->qst.pfx.byt, p+1, i->qst.pfx.len=len);
                        }
                        continue;
                case 'r': a2il(p+1, t-p-1, (int*)&i->qst.ptr.rid); continue;
                case '+': i->qst.del = 0;
                        if (CT_D == lat1ct[(unsigned char)*++p]) return p;
                        continue;
                case '-': i->qst.del = 1;
                        if (CT_D == lat1ct[(unsigned char)*++p]) return p;
                        continue;
                default:
                        if (CT_D == lat1ct[(unsigned char)*p]) return p;
                        continue;
                }
                a2il(p+1, t-p-1, (int*)&i->qst.ptr.pos);
        }
        return 0;
}       /* idxctl */


/*
        write a record.
        header must have 'W<TAB>' prestripped.
*/
static int W ( Db *db, Fld *rec )
{
        int rid = 0;
        int pos = -1;
        unsigned siz = 0;
        char *e = rec->val + rec->len;
        /* strip rid@pos */
        rec->val += a2il(rec->val, rec->len, &rid);
        if ( e > rec->val && '@' == *rec->val ) {
                rec->val++;
                rec->val += a2il(rec->val, e-rec->val, &pos);
        }
        if ( e > rec->val && TAB == *rec->val )
                rec->val++;
        rec->len = e-rec->val;
        if ( 0 < (rid = rWrite(&db->rdx, rec, rid, pos, siz)) )
                ses->rid = (unsigned)rid;
        return rid;
}       /* W */


/*
*/
int OPT_STDCALL oDbo ( Dbo *self, Msg msg )
{
        eRr(LOG_VERBOSE, "oDBo '%.*s'", msg->len, msg->val);
        if ( ! msg->len )
                goto simplewrite;
        switch (*msg->val) {
        case 'Q': {
                Qry *q;
                int i = ses->r;
                unsigned len;

                if ( 1 == msg->len ) {
                        if ( !(q = self->qry) )
                                return eRr(ERR_BADF, "query expired");
                } else if ( TAB != msg->val[1] )
                        break;
                else if ( !(q = Q(self, msg)) )
                        return ERR_INVAL;
                /* fetch results */
                len = q->res ? q->len : (unsigned)self->db->rdx.mid;
                q->use = ses->req;
                eRr(LOG_VERBOSE, "%d of %d results off %d", i, q->len, q->cur);
                eOut(0, "#\t%d\t%d\t%d", len-q->cur, q->qid, (unsigned)(q->brk>>16));
                if ( q->res )
                        while (q->cur < len && (i -= filter(self, q->res[q->cur++], q->exp)))
                                ;
                else
                        while (q->cur < len && (i -= filter(self, ++q->cur, q->exp)))
                                ;
                return 0;
        } /* Q */
        case 'R':
                if ( 1 == msg->len || TAB == msg->val[1] ) {
                        int rid, ret = 0;
                        EADD(0, "W", 1);
                        if ( 1 == msg->len ) { /* R multi */
                                Fld *end = REND(msg);
                                unsigned mr = ses->r;
                                while ( ++msg < end
                                        && 0 <= (ret = ( a2il(msg->val, msg->len, &rid),
                                                rRead(&env.out->lst, &self->db->rdx, rid, self->p)))
                                        && --mr /* allowing 4 billion if mr was 0 */
                                )
                                        SINK(env.out);
                                return ret;
                        } else { /* single or range */
                                char *p = msg->val+2, *e = p + msg->len -2;
                                int count = 1;
                                p += a2il(p, e-p, &rid);
                                if ( rid > self->db->rdx.mid )
                                        return 0;
                                if ( e > p && TAB == *p ) {
                                        p++;
                                        a2il(p, e-p, &count);
                                        if ( 0 >= count || count > (int)ses->r )
                                                count = (int)ses->r;
                                        if ( !count || count > self->db->rdx.mid - rid + 1 )
                                                count = self->db->rdx.mid - rid + 1;
                                }
                                for ( ; 0 <= (ret = rRead(&env.out->lst, &self->db->rdx, rid, self->p))
                                        && --count;
                                        rid++
                                ) {
                                        eRr(LOG_INFO, "R %u: %d lst %d", rid, ret, LLEN(&env.out->lst));
                                        SINK(env.out);
                                }
                                return ret;
                        }
                }
                break;
        case 'T':
                if ( 1 == msg->len || TAB == msg->val[1] )
                        return T(self->db, msg);
                break;
        case 'W':
                if ( 1 == msg->len ) { /* multi */
                        int left = RLEN(msg)-1, l, ret;
                        EADD(0, "R", 1);
                        for ( msg++; left; left -= l, msg += l ) {
                                if ( !(l = RLEN(msg)) )
                                        msg->tag = -(l = left);
                                else if ( l < 0 || left < l )
                                        return eRr(ERR_INVAL, "whoopsy! bad embedded rec");
                                if ( 0 > (ret = W(self->db, msg)) )
                                        return ret;
                                self->rid = (unsigned)ret;
                                eOut(0, "%d", ret);
                        }
                        return 0;
                } else if ( TAB == msg->val[1] ) { /* single */
                        msg->val += 2; /* strip W<TAB> */
                        msg->len -= 2;
                        goto simplewrite;
                }
                /* else it's not a 'W' ... */
                break;
        case '0': case '1': case '2': case '3': case '4':
        case '5': case '6': case '7': case '8': case '9':
                /* well ... we really should not see such messages ... alas */
simplewrite:
                {
                        int ret = W(self->db, msg);
                        if ( 0 < ret ) {
                                eOut(0, "R\t%d", ret);
                                self->rid = (unsigned)ret;
                        }
                        return 0;
                }
        case 'X':
                if ( 1 == msg->len || TAB == msg->val[1] ) {
                        char buf[0x2000], *mhd = buf, *val;
                        int mhdlen = sizeof(buf);
                        Fld *end = REND(msg);
                        int ctl = 1, lasttag = -1, cnt = 0, len = 0;
                        Idx idx;

                        memset(&idx, 0, sizeof(idx));
                        idx.qst.qdx = &self->db->qdx;
                        idx.qst.ptr.rid = self->rid;
                        for (;;) {
                                if ( !ctl ) {
                                        idx.qst.ptr.tag = msg->tag;
                                        val = msg->val;
                                        len = msg->len;
                                } else if ( (val = idxctl(&idx, msg)) ) {
                                        char *e = msg->val + msg->len;
                                        int tag;
                                        val += a2il(val, e-val, &tag);
                                        idx.qst.ptr.tag = (unsigned short)tag;
                                        if ( val < e && TAB == *val ) val++;
                                        len = e - val;
                                }
                                if ( val ) {
                                        if ( lasttag != idx.qst.ptr.tag ) {
                                                idx.qst.ptr.pos = 0;
                                                lasttag = idx.qst.ptr.tag;
                                        }
                                        if ( idx.mhd ) {
                                                char *ev = val+len, *d;
                                                if ( len > mhdlen ) {
                                                        if ( mhd != buf )
                                                                mFree(mhd);
                                                        mhd = mAlloc(len);
                                                }
                                                for (d = mhd;;val++) {
                                                        char *s = val;
                                                        while ( val < ev && '<' != *val ) val++;
                                                        if ( val > s ) {
                                                                memcpy(d, s, val - s);
                                                                d += val - s;
                                                        }
                                                        while ( val < ev && '>' != *val && '=' != *val ) val++;
                                                        if ( val == ev ) break;
                                                        if ( '>' == *val ) continue;
                                                        s = ++val;
                                                        while ( val < ev && '>' != *val ) val++;
                                                        if ( val > s ) {
                                                                memcpy(d, s, val - s);
                                                                d += val - s;
                                                        }
                                                }
                                                val = mhd;
                                                len = d - mhd;
                                                eRr(LOG_DEBUG, "mhd len %d", len);
                                        }
                                        if ( IDX_SPLIT == idx.mod )
                                                cnt += qSetKeyVals(&idx.qst, val, len);
                                        else {
                                                if ( !qSetKeyVal(&idx.qst, val, len) )
                                                        cnt++;
                                                if (IDX_FIELD == idx.mod)
                                                        idx.qst.ptr.pos = (idx.qst.ptr.pos + 0x10000) & OCCMSK;
                                        }
                                }
                                if ( end == ++msg )
                                        break;
                                ctl = !msg->tag;
                        }
                        if ( mhd != buf )
                                mFree(mhd);
                        eOut(0, "#\t%d", cnt);
                        return 0;
                }       /* case 'X' */
        } /* switch */
        return oStruct((Struct*)self, msg);
} /* oDbo */


extern Dbo *newDbo ( char *name )
{
        Prv *prv;
        Db *db = dOpen(name);
        if ( !db )
                return 0;
        prv = mAlloc(sizeof(Prv));
        prv->dbo.snd = (disp*)oDbo;
        prv->dbo.db = db;
        prv->dbo.qry = &prv->qry;
        prv->qry.dbo = &prv->dbo;
        prv->qry.exp = prv->qry.ebf; /* OP_NULL filter */
        return &prv->dbo;
}       /* newDbo */