#pragma prototyped

static const char merge_usage[] =

;

#include <dsslib.h>

struct File_s; typedef struct File_s File_t;

struct Key_s; typedef struct Key_s Key_t;

struct State_s; typedef struct State_s State_t;

struct File_s

{

Dtlink_t link;

Dssfile_t* file;

Dssrecord_t* record;

Cxoperand_t* data;

};

struct Key_s

{

Key_t* next;

Cxvariable_t* variable;

int sense;

};

struct State_s

{

Dtdisc_t orderdisc;

Dt_t* order;

Dss_t* dss;

Cx_t* cx;

size_t nkeys;

size_t nfiles;

Key_t* keys;

Cxcallout_f getf;

File_t files[1];

};

extern Dsslib_t dss_lib_merge;

static int

ordercmp(Dt_t* dict, void* a, void* b, Dtdisc_t* disc)

{

State_t* state = (State_t*)disc;

File_t* ap = (File_t*)a;

File_t* bp = (File_t*)b;

Key_t* kp;

size_t az;

size_t bz;

int k;

int r;

for (k = 0, kp = state->keys; k < state->nkeys; k++, kp = kp->next)

if (cxisstring(kp->variable->type) || cxisbuffer(kp->variable->type))

{

az = ap->data[k].value.buffer.size;

bz = bp->data[k].value.buffer.size;

if (r = memcmp(ap->data[k].value.buffer.data, bp->data[k].value.buffer.data, az < bz ? az : bz))

return (r < 0) ? -kp->sense : kp->sense;

if (az < bz)

return -kp->sense;

if (az > bz)

return kp->sense;

}

else if (ap->data[k].value.number < bp->data[k].value.number)

return -kp->sense;

else if (ap->data[k].value.number > bp->data[k].value.number)

return kp->sense;

if (ap < bp)

return -1;

if (ap > bp)

return 1;

return 0;

}

static void

enter(Dss_t* dss, register State_t* state, register File_t* file)

{

register int k;

register Key_t* kp;

Cxinstruction_t x;

if (file->record)

dtdelete(state->order, file);

if (file->record = dssfread(file->file))

{

for (k = 0, kp = state->keys; k < state->nkeys; k++, kp = kp->next)

{

x.data.variable = kp->variable;

if ((*state->getf)(state->cx, &x, file->data + k, NiL, NiL, file->record, state->cx->disc))

return;

}

dtinsert(state->order, file);

}

}

static int

merge_beg(Cx_t* cx, Cxexpr_t* expr, void* data, Cxdisc_t* disc)

{

register State_t* state;

register File_t* file;

char** argv = (char**)data;

char** files = expr->files;

Dss_t* dss = DSS(cx);

int errors = error_info.errors;

int i;

int k;

int n;

int r;

char* path;

char* u;

char** v;

Cxvariable_t* variable;

Cxoperand_t* operands;

Key_t* key;

Key_t* lastkey;

Key_t* keys;

Vmalloc_t* vm;

if (!(vm = vmopen(Vmdcheap, Vmbest, 0)))

{

if (disc->errorf)

(*disc->errorf)(NiL, disc, ERROR_SYSTEM|2, "out of space");

return -1;

}

r = -1;

k = 0;

keys = 0;

sfprintf(cx->buf, "%s%s", strchr(dss_lib_merge.description, '['), merge_usage);

u = sfstruse(cx->buf);

for (;;)

{

switch (i = optget(argv, u))

{

case 'k':

case 'r':

if (!(variable = cxvariable(cx, opt_info.arg, NiL, disc)))

goto bad;

if (!(key = vmnewof(vm, 0, Key_t, 1, 0)))

{

if (disc->errorf)

(*disc->errorf)(NiL, disc, ERROR_SYSTEM|2, "out of space");

goto bad;

}

key->variable = variable;

key->sense = (i == 'r') ? -1 : 1;

if (keys)

lastkey = lastkey->next = key;

else

lastkey = keys = key;

k++;

continue;

case '?':

if (disc->errorf)

(*disc->errorf)(NiL, disc, ERROR_USAGE|4, "%s", opt_info.arg);

else

return -1;

continue;

case ':':

if (disc->errorf)

(*disc->errorf)(NiL, disc, 2, "%s", opt_info.arg);

else

return -1;

continue;

}

break;

}

if (error_info.errors > errors)

goto bad;

argv += opt_info.index;

for (v = argv; *v; v++);

n = v - argv;

if (files)

{

for (v = files; *v; v++);

n += v - files;

}

if (!n)

n = 1;

if (!(state = vmnewof(vm, 0, State_t, 1, (n - 1) * sizeof(File_t))) || !(operands = vmnewof(vm, 0, Cxoperand_t, n * k, 0)))

{

if (cx->disc->errorf)

(*cx->disc->errorf)(NiL, cx->disc, ERROR_SYSTEM|2, "out of space");

goto bad;

}

state->cx = cx;

if (!(state->getf = cxcallout(cx, CX_GET, cx->state->type_void, cx->state->type_void, cx->disc)))

{

if (cx->disc->errorf)

(*cx->disc->errorf)(NiL, cx->disc, 2, "CX_GET callout required");

goto bad;

}

state->nfiles = n;

state->nkeys = k;

state->keys = keys;

for (n = 0; n < state->nfiles; n++)

{

state->files[n].data = operands;

operands += k;

}

state->orderdisc.comparf = ordercmp;

if (!(state->order = dtnew(vm, &state->orderdisc, Dtoset)))

{

if (cx->disc->errorf)

(*cx->disc->errorf)(NiL, cx->disc, ERROR_SYSTEM|2, "out of space");

goto bad;

}

n = 0;

if (path = *argv)

argv++;

else if (files)

{

argv = files;

files = 0;

if (path = *argv)

argv++;

}

for (;;)

{

if (!(state->files[n].file = dssfopen(dss, path, NiL, DSS_FILE_READ, NiL)))

goto drop;

enter(dss, state, &state->files[n]);

n++;

if (!(path = *argv++))

{

if (!files)

break;

argv = files;

files = 0;

if (!(path = *argv++))

break;

}

}

expr = expr->pass;

if (dssbeg(dss, expr))

goto drop;

while (file = (File_t*)dtfirst(state->order))

{

if (dsseval(dss, expr, file->record) < 0)

goto drop;

enter(dss, state, file);

}

if (error_info.errors == errors)

r = 0;

drop:

for (n = 0; n < state->nfiles; n++)

if (state->files[n].file)

dssfclose(state->files[n].file);

bad:

vmclose(vm);

return r;

}

static Cxquery_t queries[] =

{

{

"merge",

"merge input files",

CXH,

merge_beg,

0,

0,

0,

0,

0,

1

},

{0}

};

Dsslib_t dss_lib_merge =

{

"merge",

"merge query"

"[-1lms5P?\n@(#)$Id: dss merge query (AT&T Research) 2003-02-14 $\n]"

USAGE_LICENSE,

CXH,

0,

0,

0,

0,

0,

0,

&queries[0]

};