Martin Kersten wrote:
> Update of /cvsroot/monetdb/MonetDB5/src/modules/mal
> In directory sc8-pr-cvs7.sourceforge.net:/tmp/cvs-serv29523
>
> Modified Files:
> bpm.mx
> Log Message:
> 70% of the code needed to handle partitioned BATs in a single server.
> The code is not tested yet (except for compilation errors), but should
> give a head start for those in-experienced in working with MonetDB source code.
>
>
> Index: bpm.mx
> ===================================================================
> RCS file: /cvsroot/monetdb/MonetDB5/src/modules/mal/bpm.mx,v
> retrieving revision 1.23
> retrieving revision 1.24
> diff -u -d -r1.23 -r1.24
> --- bpm.mx 4 Feb 2007 22:55:32 -0000 1.23
> +++ bpm.mx 6 Feb 2007 07:16:13 -0000 1.24
> @@ -229,7 +229,7 @@
> address BPMdestroy
> comment "Destroy the BAT partition box";
>
> -command deposit(nme:str,b:bat[:oid,:any_2]) :void
> +command deposit(nme:str,b:bat[:oid,:any_2]) :bat[:oid,:any_2]
> address BPMdeposit
> comment "Create a new partitioned BAT by name";
>
> @@ -238,36 +238,36 @@
> If the alias BAT denotes an existing partition, it is
> further broken into pieces.
> @mal
> -command rangePartition(pb:bat[:oid,:any_2],
> - ll:oid, lh:oid, rl:any_2, rh:any_2):void
> +command rangePartition(pb:bat[:any_1,:any_2], rl:any_2, rh:any_2):void
> address BPMrange
> comment "Create a range partition on a BAT";
> -command rangePartition(pb:bat[:oid,:any_2], pv:bat[:oid,:any_2]):void
> +command rangePartition(pb:bat[:any_1,:any_2], pv:bat[:oid,:any_2]):void
> address BPMrangeVector
> comment "Create the partitions based on a range vector";
>
> -command hashPartition(pb:bat[:oid,:any_2], slots:int):void
> +command hashPartition(pb:bat[:any_1,:any_2], slots:int):void
> address BPMhash
> comment "Create a hash partition on a BAT";
> -command derivePartition(pb:bat[:oid,:any_2], src:bat[:oid,:any_2]):bat[:oid,:any_2]
> +command derivePartition(pb:bat[:any_1,:any_2],
> + src:bat[:any_1,:any_2]):bat[:any_1,:any_2]
> address BPMderived
> comment "Create a derived fragmentation over the head using src.";
>
> -command take(pb:str, b:bat[:oid,:any_2]):bat[:oid,:any_2]
> +command take(pb:str, b:bat[:any_1,:any_2]):bat[:any_1,:any_2]
> address BPMtake
> comment "Retrieve the alias for a partitioned BAT";
> -command take(alias:bat[:oid,:any_2],index:int) :bat[:oid,:any_2]
> +command take(alias:bat[:any_1,:any_2],index:int) :bat[:any_1,:any_2]
> address BPMtakePartition
> comment "Retrieve a single component of a partitioned BAT by index";
>
> -command insert(pb:bat[:oid,:any_2],b:bat[:oid,:any_2]) :void
> +command insert(pb:bat[:any_1,:any_2],b:bat[:any_1,:any_2]) :void
> address BPMinsert
> comment "Insert elements into the BAT partitions";
> -command delete(pb:bat[:oid,:any_2],b:bat[:oid,:any_2]) :void
> +command delete(pb:bat[:any_1,:any_2],b:bat[:any_1,:any_2]) :void
> address BPMdelete
> comment "Delete elements from the BAT partitions";
> -command replace(pb:bat[:oid,:any_2],old:bat[:oid,:any_2],
> - nwe:bat[:oid,:any_2]) :void
> +command replace(pb:bat[:any_1,:any_2],old:bat[:any_1,:any_2],
> + nwe:bat[:any_1,:any_2]) :void
> address BPMreplace
> comment "Replace the content of the BAT partitions";
>
> @@ -275,7 +275,7 @@
> address BPMgetNames
> comment "Retrieve the names of all known partitioned BATs";
>
> -command discard(alias:bat[:oid,:any_2]) :void
> +command discard(alias:bat[:any_1,:any_2]) :void
> address BPMdiscard
> comment "Release a partitioned BAT from the box";
>
> @@ -285,39 +285,39 @@
> Wherever possible skipping elements that don't qualify
> the bounds given for the head.
> @mal
> -command newIterator(grp:bat[:oid,:any_2]):bat[:oid,:any_2]
> +command newIterator(grp:bat[:any_1,:any_2]):bat[:any_1,:any_2]
> address BPMnewIterator
> comment "Create an iterator over the BAT partitions.";
>
> -command newIterator(grp:bat[:oid,:any_2],first:any_2,last:any_2)
> - :bat[:oid,:any_2]
> +command newIterator(grp:bat[:any_1,:any_2],first:any_2,last:any_2)
> + :bat[:any_1,:any_2]
> address BPMnewIteratorRng
> comment "Create an iterator over the BAT partitions.";
>
> -command newIterator(pb:bat[:oid,:any_2], first:oid,last:oid,
> +command newIterator(pb:bat[:any_1,:any_2], first:any_1,last:any_1,
> vlow:any_2, vhgh:any_2) :bat[:oid,:any_2]
> address BPMnewIteratorRng4
> comment "Create an iterator over the BAT partitions.";
>
> -command hasMoreElements(grp:bat[:oid,:any_2]) :bat[:oid,:any_2]
> +command hasMoreElements(grp:bat[:any_1,:any_2]) :bat[:any_1,:any_2]
> address BPMhasMoreElements
> comment "Localize the next partition for processing.";
>
> -command hasMoreElements(pb:bat[:oid,:any_2],
> - low:any_2, hgh:any_2) :bat[:oid,:any_2]
> +command hasMoreElements(pb:bat[:any_1,:any_2],
> + low:any_2, hgh:any_2) :bat[:any_1,:any_2]
> address BPMhasMoreElementsRng2
> comment "Localize the next partition for processing.";
> -command hasMoreElements(pb:bat[:oid,:any_2], first:oid,last:oid,
> - vlow:any_2, vhgh:any_2) :bat[:oid,:any_2]
> +command hasMoreElements(pb:bat[:any_1,:any_2], first:any_1,last:any_1,
> + vlow:any_2, vhgh:any_2) :bat[:any_1,:any_2]
> address BPMhasMoreElementsRng4
> comment "Localize the next partition for processing.";
>
> -command getDimension(b:bat[:oid,:any_2])(first:oid,last:oid,
> +command getDimension(b:bat[:any_1,:any_2])(first:any_1,last:any_1,
> vlow:any_2, vhgh:any_2)
> address BPMgetDimension
> comment "Obtain the partition boundary values.";
>
> -command dump(alias:bat[:oid,:any_2])
> +command dump(alias:bat[:any_1,:any_2])
> address BPMdumpAlias
> comment "Give the details of the partition tree";
> command dump()
> @@ -403,6 +403,7 @@
> @c
> #include "mal_config.h"
> #include "bpm.h"
> +#include "bat5.h"
>
> @-
> Every partition is the result of at most four operations:
> @@ -430,7 +431,6 @@
> int nxt, prv; /* list of all partitions*/
> } *Partition, PartitionRec;
>
> -/* NOT USED YET
> static void BPMprintRecord(stream *f, Partition p){
> stream_printf(f,"partition: %s alias %d bid %d index %d ",
> p->name, p->alias, p->bid, p->index);
> @@ -438,7 +438,6 @@
> stream_printf(f,"hbucket %d tbucket %d ", p->hbucket,p->tbucket);
> stream_printf(f,"nxt %d prv %d\n", p->nxt,p->prv);
> }
> -*/
> @-
> The number of partitioned BATs is considered low and
> a straight forward array with linear search seems
> @@ -470,7 +469,6 @@
> }
> return 0;
> }
> -/* NOT USED YET
> static Partition getPartition(int bid, int idx){
> return bpmcat+getPartitionIndex(bid,idx);
> }
> @@ -478,6 +476,7 @@
> return getPartition(bid,0);
> }
>
> +/* NOT USED YET
> static void delAlias(int bid){
> int i;
> i= getPartitionIndex(bid,0);
> @@ -595,33 +594,121 @@
>
> BBPkeepref(*ret= bn->batCacheid);
> BBPunfix(b->batCacheid);
> - throw(MAL, "bpm.deposit","NYI");
> + return MAL_SUCCEED;
> }
> -
> +@-
> +Range partitioning simply runs through all partitions
> +and creates new ones, keeping the partitions in order.
> +The 'critical' part is to detect overlap of the range
> +and the bounds of the partition.
> +@c
> str
> BPMrange(int *ret, int *bid, ptr *low, ptr *hgh){
> BAT *b;
> + int i, tpe;
> + int (*cmp) (ptr, ptr);
> + ptr nilptr;
> + int low_nil, hgh_nil;
> + Partition p;
> +
> BPMopen();
> - @:getBATdescriptor(bid,b,"bpm.range")@
> - /* determine the partitioning scheme */
> - (void) b;
> - (void) low;
> - (void) hgh;
> +
> + p= getAlias(*bid);
> + if( p==0){
> + throw(MAL, "bpm.range","Partition not known");
> + }
> + i= p->index;
> + @:getBATdescriptor(&bpmcat[i].bid,b,"bpm.range")@
> + if( b == 0)
> + throw(MAL,"bpm.range","Could not access BAT");
> +
> +
> + /* get the comparison function */
> + tpe= *bid >0? b->htype:b->ttype;
> + cmp= BATatoms[tpe].atomCmp;
> + nilptr = ATOMnilptr(tpe);
> + low_nil = ((*cmp) (*low, nilptr) == 0);
> + hgh_nil = ((*cmp) (*hgh, nilptr) == 0);
> +
> + BBPunfix(b->batCacheid); /* don't need it anymore */
> +
> + for( ; i ; i= bpmcat[i].nxt){
> + @:getBATdescriptor(&bpmcat[i].bid,b,"bpm.range")@
> +
> + /* determine overlag by excluding outliers */
> + if( *bid > 0 ){
> + if( (*cmp) ((ptr) &bpmcat[i].thgh, *low) < 0 ||
> + (*cmp) ((ptr) &bpmcat[i].tlow, *hgh) > 0
> + ){
> + /* this fragment need not be split */
> + } else {
> + /* break the fragment */
> +#ifdef _DEBUG_BPM_
> + stream_printf(GDKout, "break fragment %d\n",i);
> +#endif
> + }
> + } else {
> + /* use reversed bat */
> + if( (*cmp) ((ptr) &bpmcat[i].hhgh, *low) < 0 ||
> + (*cmp) ((ptr) &bpmcat[i].hlow, *hgh) > 0
> + ){
> + /* this fragment need not be split */
> + } else {
> +#ifdef _DEBUG_BPM_
> + stream_printf(GDKout, "break fragment %d\n",i);
> +#endif
> + /* break the fragment */
> + }
> + }
> + BBPunfix(b->batCacheid);
> + }
> *ret= 0;
> - throw(MAL, "bpm.range","NYI");
> + return MAL_SUCCEED;
> }
> +@-
> +The vector approach could either use a single column
> +of ordered split points, or a double column with (low,hgh)
> +value pairs. (STILL TO BE DECIDED)
> +We should ensure that the range vector is sorted first.
> +@c
> str
> BPMrangeVector(int *ret, int *bid, int *pv)
> {
> - BAT *b;
> + BAT *b, *bpv;
> + BUN p,q;
> + /* ptr ph=0; */
> +
> BPMopen();
> @:getBATdescriptor(bid,b,"bpm.range")@
> - /* determine the partitioning scheme */
> - (void) b;
> - (void) bid;
> - (void) pv;
> - *ret= 0;
> - throw(MAL, "bpm.range","NYI");
> + if( b == 0)
> + throw(MAL,"bpm.range","Cannot access BAT");
> +
> + @:getBATdescriptor(pv,bpv,"bpm.range")@
> + if( bpv == 0){
> + BBPunfix(b->batCacheid);
> + throw(MAL,"bpm.range","Cannot access BAT");
> + }
> + /* Apply the range partition using pairs */
> + BATloop(b,p,q){
> + ptr ph= BUNhead(b,p);
> + ptr pt= BUNtail(b,p);
> + BPMrange(ret, bid, &ph, &pt);
> + }
> + /* Apply the range partition using single column
> + BATloop(b,p,q){
> + if( ph){
> + ph= BUNtail(b,p);
> + } else {
> + ptr pt= BUNtail(b,p);
> + BPMrange(ret, bid,&ph,&pt);
> + ph= BUNtail(b,p);
> + }
> + }
> + */
> +
> + BBPunfix(b->batCacheid);
> + BBPunfix(bpv->batCacheid);
> + return MAL_SUCCEED;
> }
> str
> BPMhash(int *ret, int *bid, int *slots, int *prime)
> @@ -654,77 +741,140 @@
> str
> BPMtake(int *ret, str *nme)
> {
> + int i;
> BPMopen();
> - (void) nme;
> - *ret =0;
> - throw(MAL, "bpm.take","NYI");
> + i= getPartitionName(*nme);
> + if( i== 0){
> + throw(MAL,"pbm.take","Partitioned BAT does not exist");
> + }
> + *ret = bpmcat[i].alias;
> + return MAL_SUCCEED;
> }
> str
> BPMtakePartition(int *ret, int *bid, int *index)
> {
> + BAT *b;
> + int i;
> +
> BPMopen();
> - (void) bid;
> - (void) index;
> - *ret =0;
> - throw(MAL, "bpm.take","NYI");
> + i= getPartitionIndex(*bid, *index);
> + if( i== 0){
> + throw(MAL,"pbm.take","Partitioned BAT does not exist");
> + }
> + @:getBATdescriptor(bid,b,"bpm.take")@
> + if( b == 0){
> + throw(MAL,"bpm.take","Cannot access BAT");
> + }
> + BBPkeepref(*ret =b->batCacheid);
> + return MAL_SUCCEED;
> }
> @- Updates
> +The update operations simply loop through the partitions
> +and apply the results. We can optimize this by inspection
> +of the range tables, but that is considered relevant in a distributed
> +case.
> +The updates are merely convenient operators, because ideally
> +the optimizer already filters out the partitions of interest.
> @c
> str
> BPMinsert(int *ret, int *bid, int *ins)
> {
> - BAT *b;
> + int i;
> + str msg= MAL_SUCCEED;
> + Partition p;
> +
> BPMopen();
> - @:getBATdescriptor(bid,b,"bpm.insert")@
> + p= getAlias(*bid);
> + if( p== 0)
> + throw(MAL,"bpm.insert","Can not alias BAT");
> /* distributed the content */
> - (void) b;
> - (void) ins;
> + for(i= p->bid ; i && msg== MAL_SUCCEED; i= bpmcat[i].nxt){
> + msg=BKCinsert_bat(ret,&bpmcat[i].bid,ins);
> + }
> *ret= 0;
> - throw(MAL, "bpm.insert","NYI");
> + return msg;
> }
>
> str
> BPMdelete(int *ret, int *bid, int *del)
> {
> - BAT *b;
> + int i;
> + Partition p;
> + str msg= MAL_SUCCEED;
> +
> BPMopen();
> - @:getBATdescriptor(bid,b,"bpm.delete")@
> - /* delete some information from the partitions */
> - (void) b;
> - (void) del;
> + p= getAlias(*bid);
> + if( p== 0)
> + throw(MAL,"bpm.insert","Can not alias BAT");
> + /* distributed the content */
> + for(i= p->bid ; i && msg== MAL_SUCCEED; i= bpmcat[i].nxt){
> + msg=BKCdelete_bat(ret,&bpmcat[i].bid,del);
> + }
> *ret= 0;
> - throw(MAL, "bpm.delete","NYI");
> + return msg;
> }
>
> str
> BPMreplace(int *ret, int *bid, int *ins, int *del)
> {
> - BAT *b;
> + int i;
> + Partition p;
> + str msg= MAL_SUCCEED;
> +
> BPMopen();
> - @:getBATdescriptor(bid,b,"bpm.insert")@
> - /* replace values */
> - (void) b;
> - (void) ins;
> - (void) del;
> + p= getAlias(*bid);
> + if( p== 0)
> + throw(MAL,"bpm.insert","Can not alias BAT");
> + /* distributed the content */
> + for( i=p->bid ; i && msg== MAL_SUCCEED; i= bpmcat[i].nxt){
> + msg=BKCdelete_bat(ret,&bpmcat[i].bid,del);
It should be
msg=BKCdelete_bat(ret,&bpmcat[i].pid,del);
Right?
> + if( msg == MAL_SUCCEED)
> + msg=BKCinsert_bat(ret,&bpmcat[i].bid,ins);
It should be :
msg=BKCinsert_bat(ret,&bpmcat[i].pid,ins);
Rigth?
> + }
> *ret= 0;
> - throw(MAL, "bpm.replace","NYI");
> + return msg;
> }
>
> str
> BPMgetNames(int *ret)
> {
> + BAT *b;
> + int i;
> +
> BPMopen();
> - *ret =0;
> - throw(MAL, "bpm.getNames","NYI");
> + b = BATnew(TYPE_int, TYPE_str, BBPsize);
> + if (b == 0)
> + throw(MAL, "catalog.bbpNames", "failed to create BAT");
> +
> + for(i=1; i<bpmcattop; i++)
> + BUNins(b, &i, bpmcat[i].name,TRUE);
> + BBPkeepref(*ret = b->batCacheid);
> + if (!(b->batDirty&2))
> + b = BATsetaccess(b, BAT_READ);
> + return MAL_SUCCEED;
> }
>
> str
> BPMdiscard(int *ret, int *bid)
> {
> + Partition p;
> + int i;
> + signed char r;
> + str msg= MAL_SUCCEED;
> +
> BPMopen();
> - (void) bid;
> + p= getAlias(*bid);
> + if( p== 0)
> + throw(MAL,"bpm.discard","Can not alias BAT");
> + /* discard the content */
> + for(i= p->bid ; i && msg== MAL_SUCCEED; i= bpmcat[i].nxt){
> + msg= BKCdestroy(&r, &bpmcat[i].bid);
It should be:
> + msg= BKCdestroy(&ret, &bpmcat[i].bid);
Right?
If I am right I will fix and commit it. These differences created
compilation errors in windows.
Regards,
Romulo
