#pragma prototyped

static const char usage[] =

"and maxima from a record stream. Each record is categorized by the main "

"sort key. If there is no main sort key then all records are in one "

"category. If the \bmin\b key sorts less than the current category "

"minimum or if the \bmax\b key sorts greater than the category maximum "

"then the category minimum or maximum is updated and the record is "

"written to the standard output. \bmax=\b\ak1\a,\bmax=\b\ak2\a sets one "

"maximum using the two sort keys \ak1\a and \ak2\a. "

"\bmax:=\b\ak1\a,\bmax:=\b\ak2\a sets two maxima, the first using sort "

"key \ak1\a and the second using sort key \ak2\a.]"

"either be quoted or escaped \aafter\a normal shell expansion.]"

"category.]"

"category count, and total record count.]"

"{"

"[+sort -t, -k1,1 -k2,2n -lglean,flush,min='\"3,3n\"',min='\"4,4\"' old.dat -?Sorts "

"on the \b,\b separated fields 1 (string) and 2 (numeric) and "

"lists the current minimal records per field 3 (numeric) and "

"field 4 (string), for each value of the catenation of fields 1 and 2.]"

"}"

;

#include <ast.h>

#include <debug.h>

#include <dt.h>

#include <error.h>

#include <recsort.h>

#include <vmalloc.h>

typedef struct Data_s

{

void* data;

size_t size;

size_t len;

} Data_t;

typedef struct Category_s

{

Dtlink_t link;

Sfulong_t count;

unsigned int index;

Data_t key;

Data_t lim[1];

} Category_t;

struct Field_s; typedef struct Field_s Field_t;

struct Field_s

{

Field_t* next;

Rskey_t* lim;

int mm;

int index;

Category_t* category;

Sfulong_t count;

Sfulong_t total;

Data_t absolute;

};

typedef struct State_s

{

Rsdisc_t rsdisc;

Dtdisc_t dtdisc;

Rskeydisc_t kydisc;

Dt_t* categories;

unsigned int index;

Field_t* field;

unsigned int fields;

Data_t key;

Vmalloc_t* vm;

int absolute;

int count;

Category_t* all;

Sfulong_t total;

} State_t;

static int

save(Vmalloc_t* vm, register Data_t* p, void* data, size_t len)

{

if (p->size < len)

{

p->size = roundof(len, 256);

if (!(p->data = vmnewof(vm, p->data, char, p->size, 0)))

{

error(ERROR_SYSTEM|2, "out of space [save]");

return -1;

}

}

p->len = len;

if (data)

memcpy(p->data, data, len);

return 0;

}

static int

gleancmp(Dt_t* dt, void* a, void* b, Dtdisc_t* disc)

{

Category_t* x = (Category_t*)a;

Category_t* y = (Category_t*)b;

message((-4, "gleancmp a:%d:%-.*s: b:%d:%-.*s:", x->key.len, x->key.len, x->key.data, y->key.len, y->key.len, y->key.data));

if (x->key.len < y->key.len)

return -1;

if (x->key.len > y->key.len)

return 1;

return memcmp(x->key.data, y->key.data, x->key.len);

}

static char*

fmtdata(void* data, size_t size)

{

size_t i;

char* b;

char* s;

s = b = fmtbuf(2 * size + 1);

for (i = 0; i < size; i++)

s += sfsprintf(s, 3, "%02x", ((unsigned char*)data)[i]);

return b;

}

static int

glean(Rs_t* rs, int op, Void_t* data, Void_t* arg, Rsdisc_t* disc)

{

State_t* state = (State_t*)disc;

register Rsobj_t* r;

register Category_t* p;

register Field_t* f;

Category_t x;

ssize_t k;

int n;

int m;

Data_t t;

switch (op)

{

case RS_POP:

if (state->absolute)

for (f = state->field; f; f = f->next)

{

if (state->count)

sfprintf(sfstdout, "%u/%I*u/%I*u ", f->category->index, sizeof(f->count), f->count, sizeof(f->total), f->total);

sfwrite(sfstdout, f->absolute.data, f->absolute.len);

}

vmclose(state->vm);

return 0;

case RS_READ:

r = (Rsobj_t*)data;

x.key.data = r->key;

x.key.len = r->keylen;

if (state->categories && !(p = (Category_t*)dtsearch(state->categories, &x)) || !state->categories && !(p = state->all))

{

if (!(p = vmnewof(state->vm, 0, Category_t, 1, (state->fields - 1) * sizeof(Data_t) + r->keylen)))

{

error(ERROR_SYSTEM|2, "out of space [category]");

return -1;

}

p->index = ++state->index;

p->key.len = r->keylen;

p->key.data = (char*)(p + 1) + (state->fields - 1) * sizeof(Data_t);

memcpy(p->key.data, r->key, r->keylen);

if (state->categories)

dtinsert(state->categories, p);

else

state->all = p;

}

state->total++;

p->count++;

message((-2, "glean record p=%p %I*u/%I*u key='%-.*s' r:%d:%-.*s: '%-.*s'", p, sizeof(p->count), p->count, sizeof(state->total), state->total, r->keylen, r->key, x.key.len, x.key.len, x.key.data, r->datalen && r->data[r->datalen - 1] == '\n' ? r->datalen - 1 : r->datalen, r->data));

m = 0;

for (f = state->field; f; f = f->next)

{

if (f->lim->disc->defkeyf)

{

if ((k = f->lim->disc->keylen) <= 0)

k = 4;

k *= r->datalen;

if (save(state->vm, &state->key, 0, k))

return -1;

if ((k = (*f->lim->disc->defkeyf)(NiL, r->data, r->datalen, state->key.data, state->key.size, f->lim->disc)) < 0)

return -1;

t.len = state->key.len = k;

t.data = state->key.data;

}

else

{

t.data = r->data + f->lim->disc->key;

if ((k = f->lim->disc->keylen) <= 0)

k += r->datalen - f->lim->disc->key;

t.len = k;

}

message((-1, "glean [%d] %c a:%d:%s:", f->index, f->mm, t.len, fmtdata(t.data, t.len)));

if (!p->lim[f->index].data)

n = f->mm == 'm' ? -1 : 1;

else

{

message((-1, "glean [%d] %c b:%d:%s:", f->index, f->mm, p->lim[f->index].len, fmtdata(p->lim[f->index].data, p->lim[f->index].len)));

if ((k = t.len) < p->lim[f->index].len)

k = p->lim[f->index].len;

if (!(n = memcmp(t.data, p->lim[f->index].data, k)))

n = t.len - p->lim[f->index].len;

}

if (f->mm == 'm' ? (n < 0) : (n > 0))

{

if (f->lim->disc->defkeyf)

{

t = state->key;

state->key = p->lim[f->index];

p->lim[f->index] = t;

}

else if (save(state->vm, &p->lim[f->index], t.data, t.len))

return -1;

if (!state->absolute)

m = 1;

else if (save(state->vm, &f->absolute, r->data, r->datalen))

return -1;

else

{

f->category = p;

f->count = p->count;

f->total = state->total;

}

}

}

if (m)

{

if (state->count)

sfprintf(sfstdout, "%u/%I*u/%I*u ", p->index, sizeof(p->count), p->count, sizeof(state->total), state->total);

sfwrite(sfstdout, r->data, r->datalen);

}

return RS_DELETE;

}

return -1;

}

Rsdisc_t*

rs_disc(Rskey_t* key, const char* options)

{

register State_t* state;

register Field_t* f;

Vmalloc_t* vm;

int i;

if (!(vm = vmopen(Vmdcheap, Vmbest, 0)) || !(state = vmnewof(vm, 0, State_t, 1, 0)))

error(ERROR_SYSTEM|3, "out of space [state]");

state->vm = vm;

key->type &= ~RS_DATA;

state->dtdisc.link = offsetof(Category_t, link);

state->dtdisc.comparf = gleancmp;

state->kydisc.version = RSKEY_VERSION;

state->kydisc.errorf = errorf;

state->rsdisc.eventf = glean;

state->rsdisc.events = RS_READ|RS_POP;

if ((key->keydisc->flags & RSKEY_KEYS) && !(state->categories = dtnew(vm, &state->dtdisc, Dtoset)))

error(ERROR_SYSTEM|3, "out of space [dictionary]");

if (options)

{

for (;;)

{

switch (i = optstr(options, usage))

{

case 0:

break;

case 'a':

state->absolute = opt_info.num;

continue;

case 'c':

state->count = opt_info.num;

continue;

case 'f':

sfset(sfstdout, SF_LINE, 1);

continue;

case 'm':

case 'M':

if (opt_info.assignment == ':' || !(f = state->field) || f->mm != i)

{

if (!(f = vmnewof(vm, 0, Field_t, 1, 0)) || !(f->lim = rskeyopen(&state->kydisc, NiL)))

error(ERROR_SYSTEM|3, "out of space");

strcpy((char*)f->lim->tab, (char*)key->tab);

f->lim->type = key->type;

f->mm = i;

f->index = state->fields++;

f->next = state->field;

state->field = f;

}

if (rskey(f->lim, opt_info.arg, 0))

goto drop;

continue;

case '?':

error(ERROR_USAGE|4, "%s", opt_info.arg);

goto drop;

case ':':

error(2, "%s", opt_info.arg);

goto drop;

}

break;

}

for (f = state->field; f; f = f->next)

if (rskeyinit(f->lim))

goto drop;

}

return &state->rsdisc;

drop:

for (f = state->field; f; f = f->next)

rskeyclose(f->lim);

vmclose(vm);

return 0;

}

SORTLIB(glean)