/*
 * The routines @emph{ALIGN_synced} and @emph{ALIGN_ordered} allow to
 * simply query the alignment status of the two head columns of two
 * BATs.
 */
int
ALIGNsynced(BAT *b1, BAT *b2)
{
	BATcheck(b1, "ALIGNsynced: bat 1 required", 0);
	BATcheck(b2, "ALIGNsynced: bat 2 required", 0);

	/* first try to prove head columns are not in sync */
	if (BATcount(b1) != BATcount(b2))
		return 0;
	if (ATOMtype(BAThtype(b1)) != ATOMtype(BAThtype(b2)))
		return 0;
	if (BAThvoid(b1) && BAThvoid(b2))
		return (b1->hseqbase == b2->hseqbase);

	/* then try that they are */
	if (b1->batCacheid == b2->batCacheid)
		return 1;	/* same bat. trivial case */
	if (BATcount(b1) == 0)
		return 1;	/* empty bats of same type. trivial case */
	if (b1->halign && b1->halign == b2->halign)
		return 1;	/* columns marked as equal by algorithmics */
	if (VIEWparentcol(b1) && ALIGNsynced(BBPcache(VIEWhparent(b1)), b2))
		return 1;	/* view on same bat --- left recursive def.. */
	if (VIEWparentcol(b2) && ALIGNsynced(b1, BBPcache(VIEWhparent(b2))))
		return 1;	/* view on same bat --- right recursive def.. */

	return 0;		/* we simply don't know */
}

static AGGRtask*
GROUPcollect( Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci){
	AGGRtask *a;
	int i;
	BAT *b, *bs, *bh = NULL;
	BUN sample;

	(void) mb;
	(void) cntxt;
	a= (AGGRtask *) GDKzalloc(sizeof(*a));
	if ( a == NULL)
		return NULL;
	a->bid = (bat*) GDKzalloc(pci->argc * sizeof(bat));
	a->cols = (BAT**) GDKzalloc(pci->argc * sizeof(BAT*));
	a->unique = (BUN *) GDKzalloc(pci->argc * sizeof(BUN));
	if ( a->cols == NULL || a->bid == NULL || a->unique == NULL){
		if(a->cols) GDKfree(a->cols);
		if(a->bid) GDKfree(a->bid);
		if(a->unique) GDKfree(a->unique);
		GDKfree(a);
		return NULL;
	}
	for ( i= pci->retc; i< pci->argc; i++, a->last++) {
		a->bid[a->last] = *getArgReference_bat(stk,pci,i);
		b = a->cols[a->last]= BATdescriptor(a->bid[a->last]);
		if ( a->cols[a->last] == NULL){
			for(a->last--; a->last>=0; a->last--)
				BBPunfix(a->cols[a->last]->batCacheid);
			GDKfree(a->cols);
			GDKfree(a->bid);
			GDKfree(a->unique);
			GDKfree(a);
			return NULL;
		}
		sample = BATcount(b) < 1000 ? BATcount(b): 1000;
		bs = BATsample( b, sample);
		if (bs) {
			bh = BATunique(b, bs);
			if (bh) {
				a->unique[a->last] = BATcount(bh);
				BBPunfix(bh->batCacheid);
			}
			BBPunfix(bs->batCacheid);
		}
		if ( b->tsorted)
			a->unique[a->last] = 1000; /* sorting helps grouping */
		a->size = BATcount(b);
	}

#ifdef _DEBUG_GROUPBY_
	for(i=0; i<a->last; i++)
		fprintf(stderr,"#group %d unique "BUNFMT "\n", i, a->unique[i]);
#endif
	return a;
}

str DCselectInsert(int *ret, int *res, int *bid, lng *low, lng *hgh)
{
	BAT *b, *r;

	lng *readerH, *writerH;
	lng *readerT, *writerT;

	BUN size, i;

	(void) ret;

	if ((b = BATdescriptor(*bid)) == NULL)
		throw(MAL, "dc.selectInsert", "Cannot access input BAT");

	if ((r = BATdescriptor(*res)) == NULL)
		throw(MAL, "dc.selectInsert", "Cannot access result BAT");

	size = BATcount(b);

	if (size > BATcapacity(r) - BATcount(r)) {
		BUN ncap;
		BUN grows;
		BUN needed = size - (BATcapacity(r) - BATcount(r));
		ncap = BATcapacity(r) + needed;
		grows = BATgrows(r);
		if (ncap > grows)
			grows = ncap;
		if (BATextend(r, grows) == NULL)
			throw(MAL, "dc.selectInsert", "Failed to make room for the new values");
	}
/*printf("in dc.selectInsert size is "OIDFMT,size);*/
	writerH = (lng *) Hloc(r, BUNfirst(r));
	writerT = (lng *) Tloc(r, BUNfirst(r));

	readerH = (lng *) Hloc(b, BUNfirst(b));
	readerT = (lng *) Tloc(b, BUNfirst(b));

	for (i = 0; i < size; i++) {
		if (*readerT >= *low && *readerT <= *hgh) {
			*writerH = *readerH;
			*writerT = *readerT;

			writerH++;
			writerT++;
		}
		readerH++;
		readerT++;
	}
	BATsetcount(r, (BUN) (writerT - (lng *) Tloc(r, BUNfirst(r))));

	BBPunfix(*bid);
	BBPunfix(*res);

	return MAL_SUCCEED;
}

/*
 * Enable incremental packing. The SQL front-end requires
 * fixed oid sequences.
 */
str
MATpackIncrement(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr p)
{
	bat *ret = getArgReference_bat(stk,p,0);
	int	pieces;
	BAT *b, *bb, *bn;
	size_t newsize;

	(void) cntxt;
	b = BATdescriptor( stk->stk[getArg(p,1)].val.ival);
	if ( b == NULL)
		throw(MAL, "mat.pack", RUNTIME_OBJECT_MISSING);

	if ( getArgType(mb,p,2) == TYPE_int){
		/* first step, estimate with some slack */
		pieces = stk->stk[getArg(p,2)].val.ival;
		bn = BATnew(TYPE_void, b->ttype?b->ttype:TYPE_oid, (BUN)(1.2 * BATcount(b) * pieces), TRANSIENT);
		if (bn == NULL)
			throw(MAL, "mat.pack", MAL_MALLOC_FAIL);
		/* allocate enough space for the vheap, but not for strings,
		 * since BATappend does clever things for strings */
		if ( b->T->vheap && bn->T->vheap && ATOMstorage(b->ttype) != TYPE_str){
			newsize =  b->T->vheap->size * pieces;
			if (HEAPextend(bn->T->vheap, newsize, TRUE) != GDK_SUCCEED)
				throw(MAL, "mat.pack", MAL_MALLOC_FAIL);
		}
		BATseqbase(bn, b->H->seq);
		BATseqbase(BATmirror(bn), b->T->seq);
		BATappend(bn,b,FALSE);
		assert(!bn->H->nil || !bn->H->nonil);
		assert(!bn->T->nil || !bn->T->nonil);
		bn->H->align = (pieces-1);
		BBPkeepref(*ret = bn->batCacheid);
		BBPunfix(b->batCacheid);
	} else {
		/* remaining steps */
		bb = BATdescriptor(stk->stk[getArg(p,2)].val.ival);
		if ( bb ){
			if (BATcount(b) == 0)
				BATseqbase(b, bb->H->seq);
			if (BATcount(b) == 0)
				BATseqbase(BATmirror(b), bb->T->seq);
			BATappend(b,bb,FALSE);
		}
		b->H->align--;
		if(b->H->align == 0)
			BATsetaccess(b, BAT_READ);
		assert(!b->H->nil || !b->H->nonil);
		assert(!b->T->nil || !b->T->nonil);
		BBPkeepref(*ret = b->batCacheid);
		if( bb) 
			BBPunfix(bb->batCacheid);
	}
	return MAL_SUCCEED;
}

/*
 * @-
 * The operator below is only working for a very limited cases.
 */
str DCreplaceTailBasedOnHead(int *ret, int *res, int *bid)
{
	BAT *b, *r;
	oid *readerH_b;
	int *writerT_r, *readerT_b;
	BUN size_b, size_r, i;

	(void) ret;

	if ((b = BATdescriptor(*bid)) == NULL)
		throw(MAL, "dc.replaceTailBasedOnHead", "Cannot access input BAT");
	/* check for a failure */
	assert(b != NULL);

	if ((r = BATdescriptor(*res)) == NULL)
		throw(MAL, "dc.replaceTailBasedOnHead", "Cannot access result BAT");
	/* check for a failure */
	assert(r != NULL);


	/* remove Hashes etc */
	if (r->H->hash)
		HASHremove(r);
	if (r->T->hash)
		HASHremove(BATmirror(r));

	size_r = BATcount(r);
	size_b = BATcount(b);


	if ((b->htype == TYPE_void) && (size_b == size_r)) {
		writerT_r = (int *) Tloc(r, BUNfirst(r));
		readerT_b = (int *) Tloc(b, BUNfirst(b));
		for (i = 0; i < size_r; i++) {
			*writerT_r = *readerT_b;
			writerT_r++;
			readerT_b++;
		}
	} else if ((b->htype != TYPE_void) && (size_b < size_r)) {
		readerH_b = (oid *) Hloc(b, BUNfirst(b));
		readerT_b = (int *) Tloc(b, BUNfirst(b));
		for (i = 0; i < size_b; i++) {
			writerT_r = (int *) Tloc(r, BUNfirst(r)) + *readerH_b;
			*writerT_r = *readerT_b;
			readerH_b++;
			readerT_b++;
		}
	}

	BBPunfix(*bid);
	BBPunfix(*res);
	r->batDirty = TRUE;
	return MAL_SUCCEED;
}

static BAT *
MATsort_bte( BAT **map, BAT **bats, int len, BUN cnt, int rev )
{
	BAT *res;
	int i;
	bte *resT, **batsT, *in;
	bte *mapT;
	BUN len1, len2;
	bte *map_in = NULL;

	res = BATnew(TYPE_void, bats[0]->ttype, cnt, TRANSIENT);
	*map = BATnew(TYPE_void, TYPE_bte, cnt, TRANSIENT);
	if (res == NULL || *map == NULL) {
		BBPreclaim(res);
		BBPreclaim(*map);
		*map = NULL;
		return NULL;
	}
	BATseqbase(res, 0);
	BATseqbase(*map, 0);
	resT = (bte*)Tloc(res, 0);
	mapT = (bte*)Tloc(*map, 0);
	batsT = (bte**)GDKmalloc(sizeof(bte*) * len);
	for (i=0; i<len; i++)
		batsT[i] = (bte*)Tloc(bats[i], 0);
	/* merge */
	in = batsT[0];
	len1 = BATcount(bats[0]);
	map_in = NULL;
	/* TODO: change into a tree version */
	for (i=1; i<len; i++) {
		len2 = BATcount(bats[i]);
		if (rev) {
			MATsortloop_bte_rev( resT+cnt-len1-len2, 
					mapT+cnt-len1-len2, 
				        in, map_in, len1, 
					batsT[i], i, len2);
		} else {
			MATsortloop_bte_( resT+cnt-len1-len2, 
					mapT+cnt-len1-len2, 
				        in, map_in, len1, 
					batsT[i], i, len2);
		}
		in = resT+cnt-len1-len2;
		map_in = mapT+cnt-len1-len2;
		len1 += len2;
	}
	BATsetcount(res, len1);
	BATsetcount(*map, len1);
	res->hrevsorted = len1 <= 1;
	(*map)->hrevsorted = len1 <= 1;
	GDKfree(batsT);
	return res;
}

/*
 * The pack is an ordinary multi BAT insert. Oid synchronistion
 * between pieces should be ensured by the code generators.
 * The pack operation could be quite expensive, because it
 * may create a really large BAT.
 * The slice over a mat helps to avoid constructing intermediates
 * that are subsequently reduced.
 * Contrary to most operations, NIL arguments are skipped and
 * do not produce RUNTIME_OBJECT_MISSING.
 */
static str
MATpackInternal(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr p)
{
	int i, *ret = (int*) getArgReference(stk,p,0);
	BAT *b, *bn;
	BUN cap = 0;
	int tt = TYPE_any;
	(void) cntxt;
	(void) mb;

	for (i = 1; i < p->argc; i++) {
		int bid = stk->stk[getArg(p,i)].val.ival;
		b = BBPquickdesc(abs(bid),FALSE);
		if (b && bid < 0)
			b = BATmirror(b);
		if( b ){
			assert(BAThdense(b));
			if (tt == TYPE_any){
				tt = b->ttype;
			}
			if (!tt && tt != b->ttype)
				tt = b->ttype;
			cap += BATcount(b);
		}
	}
	if (tt == TYPE_any){
		*ret = 0;
		return MAL_SUCCEED;
	}

	bn = BATnew(TYPE_void, tt, cap, TRANSIENT);
	if (bn == NULL)
		throw(MAL, "mat.pack", MAL_MALLOC_FAIL);

	for (i = 1; i < p->argc; i++) {
		b = BATdescriptor(stk->stk[getArg(p,i)].val.ival);
		if( b ){
			if (BATcount(bn) == 0)
				BATseqbase(bn, b->H->seq);
			if (BATcount(bn) == 0)
				BATseqbase(BATmirror(bn), b->T->seq);
			BATappend(bn,b,FALSE);
			BBPunfix(b->batCacheid);
		}
	}
	assert(!bn->H->nil || !bn->H->nonil);
	assert(!bn->T->nil || !bn->T->nonil);
	BATsettrivprop(bn);
	BATderiveProps(bn,FALSE);
	BBPkeepref(*ret = bn->batCacheid);
	return MAL_SUCCEED;
}

static BAT *
delta_full_bat_( sql_column *c, sql_delta *bat, int temp, BAT *d, BAT *s)
{
	/* return full normalized column bat

		if (s) {
			b := b.semijoin(s);
			i := i.semijoin(s);
			u := u.semijoin(s);
		}
		b := b.kunion(i);
		b := b.kdiff(u);
		b := b.kunion(u);
		b := b.kdiff(reverse(d));
	*/
	BAT *r, *b, *u, *i = temp_descriptor(bat->ibid);
	r = i; 
	if (temp) {
		if (s) {
			r = BATsemijoin(i,s);
			bat_destroy(i);
		}
		return r;
	}
	b = temp_descriptor(bat->bid);
	u = temp_descriptor(bat->ubid);
	if (s) {
		BAT *t;

		t = BATsemijoin(b,s); bat_destroy(b); b = t;
		t = BATsemijoin(i,s); bat_destroy(i); i = t;
		t = BATsemijoin(u,s); bat_destroy(u); u = t;
	}
	assert(b->ttype == i->ttype);
	if (BATcount(i)) {
		r = BATkunion(b,i); bat_destroy(b); b = r;
	}
	bat_destroy(i); 
	if (BATcount(u)) {
		r = BATkdiff(b,u); bat_destroy(b); b = r;
		assert(b->ttype == u->ttype);
		r = BATkunion(b,u); bat_destroy(b); b = r;
	}
	bat_destroy(u); 
	if (d && BATcount(d)) {
		r = BATkdiff(b,BATmirror(d)); bat_destroy(b); b = r;
	}
	if (!bat->cached && !c->base.wtime && !s) 
		bat->cached = temp_descriptor(b->batCacheid);
	return b;
}

str
db_password_wrap(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
{
	(void) mb;

	if (stk->stk[pci->argv[0]].vtype == TYPE_bat) {
		BAT *b = BATdescriptor(*getArgReference_bat(stk, pci, 1));
		if (b == NULL)
			throw(SQL, "sql.password", SQLSTATE(HY002) RUNTIME_OBJECT_MISSING);
		BAT *bn = COLnew(b->hseqbase, TYPE_str, BATcount(b), TRANSIENT);
		if (bn == NULL) {
			BBPunfix(b->batCacheid);
			throw(SQL, "sql.password", SQLSTATE(HY001) MAL_MALLOC_FAIL);
		}
		BATiter bi = bat_iterator(b);
		BUN p, q;
		BATloop(b, p, q) {
			char *hash, *msg;
			msg = AUTHgetPasswordHash(&hash, cntxt, BUNtvar(bi, p));
			if (msg != MAL_SUCCEED) {
				BBPunfix(b->batCacheid);
				BBPreclaim(bn);
				return msg;
			}
			if (BUNappend(bn, hash, false) != GDK_SUCCEED) {
				BBPunfix(b->batCacheid);
				BBPreclaim(bn);
				throw(SQL, "sql.password", SQLSTATE(HY001) MAL_MALLOC_FAIL);
			}
			GDKfree(hash);
		}

str
CMDbbpCount(bat *ret)
{
	BAT *b, *bn;
	int i;
	lng l;

	b = BATnew(TYPE_void, TYPE_lng, getBBPsize(), TRANSIENT);
	if (b == 0)
		throw(MAL, "catalog.bbpCount", MAL_MALLOC_FAIL);
	BATseqbase(b,0);

	for (i = 1; i < getBBPsize(); i++)
		if (i != b->batCacheid) {
			if (BBP_logical(i) && (BBP_refs(i) || BBP_lrefs(i))) {
				bn = BATdescriptor(i);
				if (bn) {
					l = BATcount(bn);
					BUNappend(b,  &l, FALSE);
					BBPunfix(bn->batCacheid);
				}
			}
		}
	if (!(b->batDirty&2)) BATsetaccess(b, BAT_READ);
	pseudo(ret,b,"bbp","count");
	return MAL_SUCCEED;
}

str
batstr_2time_timestamptz(bat *res, const bat *bid, const int *digits, int *tz)
{
	BAT *b, *dst;
	BATiter bi;
	BUN p, q;
	char *msg = NULL;

	if ((b = BATdescriptor(*bid)) == NULL) {
		throw(SQL, "batcalc.str_2time_timestamp", "Cannot access descriptor");
	}
	bi = bat_iterator(b);
	dst = BATnew(TYPE_void, TYPE_timestamp, BATcount(b), TRANSIENT);
	if (dst == NULL) {
		BBPunfix(b->batCacheid);
		throw(SQL, "sql.timestamp", MAL_MALLOC_FAIL);
	}
	BATseqbase(dst, b->hseqbase);
	BATloop(b, p, q) {
		char *v = (char *) BUNtail(bi, p);
		union {
			lng l;
			timestamp r;
		} u;
		msg = str_2time_timestamptz(&u.r, &v, digits, tz);
		if (msg) {
			BBPunfix(dst->batCacheid);
			BBPunfix(b->batCacheid);
			return msg;
		}
		BUNappend(dst, &u.r, FALSE);
	}

static str
gsl_bat_chisqprob_cst(bat * retval, bat chi2, dbl datapoints) 
{
	BAT *b, *bn;
	BATiter bi;
	BUN p,q;
	dbl r;
	char *msg = NULL;

	if (datapoints == dbl_nil) {
		throw(MAL, "GSLbat_chisqprob_cst", "Parameter datapoints should not be nil");
	}
	if (datapoints < 0)
		throw(MAL, "gsl.chi2prob", "Wrong value for datapoints");

	if ((b = BATdescriptor(chi2)) == NULL) {
		throw(MAL, "chisqprob", "Cannot access descriptor");
	}
	bi = bat_iterator(b);
	bn = BATnew(TYPE_void, TYPE_dbl, BATcount(b), TRANSIENT);
	if (bn == NULL){
		BBPunfix(b->batCacheid);
		throw(MAL, "gsl.chisqprob", MAL_MALLOC_FAIL);
	}
	BATseqbase(bn, b->hseqbase);
	BATloop(b,p,q) {
		dbl d = *(dbl*)BUNtail(bi,p);
		if ((d == dbl_nil) || (d < 0))
			throw(MAL, "gsl.chi2prob", "Wrong value for chi2");
		r = gsl_cdf_chisq_Q(d, datapoints);
		BUNappend(bn, &r, FALSE);
	}

static BAT *
MATproject_hge( BAT *map, BAT **bats, int len, int ttpe )
{
	BAT *res;
	int i;
	BUN j, cnt = BATcount(map);
	hge *resT, **batsT;
	bte *mapT;

	res = BATnew(TYPE_void, ttpe, cnt, TRANSIENT);
	batsT = (hge**)GDKmalloc(sizeof(hge*) * len);
	if (res == NULL || batsT == NULL) {
		if (res)
			BBPreclaim(res);
		if (batsT)
			GDKfree(batsT);
		return NULL;
	}
	BATseqbase(res, map->hseqbase);
	resT = (hge*)Tloc(res, 0);
	mapT = (bte*)Tloc(map, 0);
	for (i=0; i<len; i++)
		batsT[i] = (hge*)Tloc(bats[i], 0);
	for (j=0; j<cnt; j++)
		resT[j] = *batsT[mapT[j]]++;
	BATsetcount(res, j);
	res->hrevsorted = j <= 1;
	GDKfree(batsT);
	return res;
}

/*
 * The nextChunk version advances the reader,
 * which also means that the view descriptor is already available.
 * The granule size may differ in each call.
 */
str
ITRnextChunk(lng *res, int *vid, int *bid, lng *granule)
{
	BAT *b, *view;
	BUN i;

	if ((b = BATdescriptor(*bid)) == NULL) {
			throw(MAL, "iterator.nextChunk", INTERNAL_BAT_ACCESS);
	}
	if ((view = BATdescriptor(*vid)) == NULL) {
		BBPunfix(b->batCacheid);
		throw(MAL, "iterator.nextChunk", INTERNAL_BAT_ACCESS);
	}
	i = (BUN) (*res + BATcount(view));
	if (i >= BUNlast(b)) {
		*res = lng_nil;
		*vid = 0;
		BBPunfix(view->batCacheid);
		BBPunfix(b->batCacheid);
		return MAL_SUCCEED;
	}
	/* printf("set bat chunk bound to " BUNFMT " - " BUNFMT " \n",
	   i, i+(BUN) *granule-1); */
	VIEWbounds(b, view, i, i + (BUN) * granule);
	BATseqbase(view, b->hseqbase == oid_nil ? oid_nil : b->hseqbase + i - BUNfirst(b));
	BBPkeepref(*vid = view->batCacheid);
	BBPunfix(b->batCacheid);
	*res = i;
	return MAL_SUCCEED;
}

static void *
column_find_value(sql_trans *tr, sql_column *c, oid rid)
{
	BUN q = BUN_NONE;
	BAT *b;
	void *res = NULL;

	b = full_column(tr, c);
	if (b) {
		if (rid < b->hseqbase || rid >= b->hseqbase + BATcount(b))
			q = BUN_NONE;
		else
			q = rid - b->hseqbase;
	}
	if (q != BUN_NONE) {
		BATiter bi = bat_iterator(b);
		const void *r;
		size_t sz;

		r = BUNtail(bi, q);
		sz = ATOMlen(b->ttype, r);
		res = GDKmalloc(sz);
		if (res)
			memcpy(res, r, sz);
	}
	full_destroy(c, b);
	return res;
}

str
DCsliceStrict(int *ret, bat *bid, lng *start, lng *end)
{
	BAT *b, *bn = NULL;

	if ((b = BATdescriptor(*bid)) == NULL) {
		throw(MAL, "dcoperator.sliceStrict", "Cannot access descriptor");
	}

	assert(*start >= 0);
	assert(*end >= 0);
	assert(*start <= (lng) BUN_MAX);
	assert(*end < (lng) BUN_MAX);
	assert(*start <= *end);

	if ((BUN) ((*end - *start) + 1) > BATcount(b)) {
		bn = BATnew(b->htype, b->ttype, 0);
		BATsetcount(bn, 0);
		*ret = bn->batCacheid;
		BBPkeepref(*ret);
		return MAL_SUCCEED;
	}

	bn = BATslice(b, (BUN) *start, (BUN) *end + 1);

	BBPreleaseref(b->batCacheid);
	if (bn != NULL) {
		if (!(bn->batDirty & 2))
			bn = BATsetaccess(bn, BAT_READ);
		*ret = bn->batCacheid;
		BBPkeepref(*ret);
		return MAL_SUCCEED;
	}
	throw(MAL, "dcoperator.sliceStrict", "GDKerror");
}

/*
 * @-
 * The BUN- and BAT-stream manipulate a long handle, i.e.
 * the destination variable. It assumes it has been set to
 * zero as part of runtime stack initialization. Subsequently,
 * it fetches a bun and returns the increment to the control
 * variable. If it returns zero the control variable has been reset
 * to zero and end of stream has been reached.
 */
str
ITRbunIterator(Client cntxt, MalBlkPtr mb, MalStkPtr stk, InstrPtr pci)
{
	BATiter bi;
	BAT *b;
	oid *head;
	int *bid;
	ValPtr tail;

	(void) cntxt;
	(void) mb;
	head = (oid *) getArgReference(stk, pci, 0);
	tail = getArgReference(stk,pci,1);
	bid = (int *) getArgReference(stk, pci, 2);

	if ((b = BATdescriptor(*bid)) == NULL) {
		throw(MAL, "iterator.nextChunk", INTERNAL_BAT_ACCESS);
	}

	if (BATcount(b) == 0) {
		*head = oid_nil;
		BBPunfix(b->batCacheid);
		return MAL_SUCCEED;
	}
	*head = BUNfirst(b);

 	bi = bat_iterator(b);
	VALinit(tail, b->ttype, BUNtail(bi, *(BUN*) head));
	BBPunfix(b->batCacheid);
	return MAL_SUCCEED;
}

static BAT *
MATproject_any( BAT *map, BAT **bats, int len )
{
	BAT *res;
	int i;
	BUN j, cnt = BATcount(map);
	BATiter *bats_i;
	BUN *batsT;
	bte *mapT;

	res = BATnew(TYPE_void, bats[0]->ttype, cnt, TRANSIENT);
	batsT = (BUN*)GDKmalloc(sizeof(BUN) * len);
	bats_i = (BATiter*)GDKmalloc(sizeof(BATiter) * len);
	if (res == NULL || batsT == NULL || bats_i == NULL) {
		if (res)
			BBPreclaim(res);
		if (batsT)
			GDKfree(batsT);
		if (bats_i)
			GDKfree(bats_i);
		return NULL;
	}
	BATseqbase(res, map->hseqbase);
	mapT = (bte*)Tloc(map, 0);
	for (i=0; i<len; i++) {
		batsT[i] = 0;
		bats_i[i] = bat_iterator(bats[i]);
	}
	for (j=0; j<cnt; j++)
		BUNappend(res, BUNtail(bats_i[mapT[j]], batsT[mapT[j]]++), FALSE);
	GDKfree(batsT);
	GDKfree(bats_i);
	return res;
}

/*
 * @-
 * The operator below is only working for a very limited
 * case. It also re-uses oids, which may become a semantic
 * problem quickly.
 */
str DCdeleteUpperSlice(int *ret, int *bid, int *pos)
{
	BAT *b;
	int *readerT, *writerT;
	BUN size, i;

	(void) ret;

	if ((b = BATdescriptor(*bid)) == NULL)
		throw(MAL, "dc.deleteUpperSlice", "Cannot access input BAT");
	/* check for a failure */
	assert(b != NULL);

	/* remove Hashes etc */
	if (b->H->hash)
		HASHremove(b);
	if (b->T->hash)
		HASHremove(BATmirror(b));

	size = BATcount(b);
	writerT = (int *) Tloc(b, BUNfirst(b));
	readerT = (int *) Tloc(b, BUNfirst(b)) + *pos;

	for (i = *pos; i < size; i++)
		*writerT++ = *readerT++;
	b->batInserted -= *pos;

	BATsetcount(b, (BUN) (writerT - (int *) Tloc(b, BUNfirst(b))));
	BBPunfix(*bid);
	b->batDirty = TRUE;
	return MAL_SUCCEED;
}

log_bid 
ebat_copy(log_bid b, oid ibase, int temp)
{
	/* make a copy of b */
	BAT *o = temp_descriptor(b);
	BAT *c;
	log_bid r;

	if (!ebats[o->ttype]) 
		ebats[o->ttype] = bat_new(TYPE_void, o->ttype, 0);

	if (!temp && BATcount(o)) {
		c = BATcopy(o, TYPE_void, o->ttype, TRUE);
		BATseqbase(c, ibase );
		c->H->dense = 1;
		BATcommit(o);
		BATcommit(c);
		bat_set_access(c, BAT_READ);
		r = temp_create(c);
		bat_destroy(c);
	} else {
		c = ebats[o->ttype];
		r = temp_create(c);
	}
	bat_destroy(o);
	return r;
}