bpm.mx

一个内存数据库的源代码这是服务器端还有客户端

@' The contents of this file are subject to the MonetDB Public License
@' Version 1.1 (the "License"); you may not use this file except in
@' compliance with the License. You may obtain a copy of the License at
@' http://monetdb.cwi.nl/Legal/MonetDBLicense-1.1.html
@'
@' Software distributed under the License is distributed on an "AS IS"
@' basis, WITHOUT WARRANTY OF ANY KIND, either express or implied. See the
@' License for the specific language governing rights and limitations
@' under the License.
@'
@' The Original Code is the MonetDB Database System.
@'
@' The Initial Developer of the Original Code is CWI.
@' Portions created by CWI are Copyright (C) 1997-2007 CWI.
@' All Rights Reserved.

@f bpm
@v 0.3
@a M.L.Kersten
@* BAT Partition Manager
In real-life database applications the BATs tend to grow beyond
the memory size. This leads to a heavy IO dominated behavior,
which can partly be avoided by breaking up the query into a sequence
of subqueries using a map-reduce strategy.
The BAT partition manager (BPM) module is designed to 
support this strategy using range- and hash-partitioning.

Consider we want to reorganize R:bat[:oid,:int] into three range partitions,
based on splitting on the head. Two partitions are stable
and the third partition is reserved for appends.
The following MAL program illustrates the snippet of actions needed:
@example
	bpm.open();
	bpm.deposit("myversion",R);	# creates side effects
	Ralias:= bpm.take("myversion",:bat[:oid,:int]);
	bpm.rangePartition(Ralias,nil:oid,100:oid,nil:int,nil:int);
	bpm.rangePartition(Ralias,101:oid,200:oid,nil:int,nil:int);
	bpm.rangePartition(Ralias,201:oid,nil:oid,nil:int,nil:int);
	bpm.close();
@end example
The command @sc{bpm.deposit} registers a BAT as one to be partitioned.
The side effect is that R is empty after this call.
The remainder are the partition definitions based
on slices from the table using a simple range condition.
The BAT partitions share the persistency properties.

Beware that now there are four components R, R1, R2, and R3.
The BAT R is is turned into an empty bat, its content is
taken over by the @sc{bpm} and broken into pieces.
The partitioned version is referenced with a BAT variable
initialized with the @sc{take} operation.
The call @sc{bpm.discard(Ralias)} removes all partitions.

The partition manager also supports hash-based partitioning.
It accepts two arguments, which denote the number of hash buckets
on the head and tail respectively.
@example
	bpm.open();
	bpm.deposit("myHashVersion",R);	# creates side effects
	Ralias:= bpm.take("myHashVersion",:bat[:oid,:int]);
	bpm.hashPartition(Ralias,5,2);
	bpm.close();
@end example
This example leads to 10 partitions.
Re-partitioning automatically occurs when a new hash size is given.

It should not come as a surprise, that a combination of
hashing and range is also provided using the
commands @sc{bpm.hashRangePartition()} and @sc{bpm.rangeHashPartition()}.

The design is based on the assumption that partitions
are reasonably large. This helps to limit plan explosion.
(or a scheduler should step in)

@- Derived partitioning
A relational front-end would benefit from derived horizontal
fragmentation. It would enable grouping together related
fragments on the same site. 
Assume a relation R(A,B) which is already partitioned on A
the derived fragmentation on the oid head is enforced with
@example
bpm.derivedPartition(B,A);
@end example
@- Using partitions
The partitioned BAT can be used in two ways. A query plan can be
rewritten into a generator over the partitions, or it can be
used by optimizers to derived all subqueries first for
symbolic evaluation.

The former is illustrated with the snippet to select part of
a partitioned BAT. In this example we collect the partial
results in the accumulator BAT tu.
@example
	bpm.open();
	R:= bpm.take("myversion",:bat[:oid,:int]);
	tu:= bat.new(:oid,:int);
barrier (idx,Rp):= bpm.newIterator(R);
	...
	t:= algebra.select(Rp,0,100);
	tu:= algebra.union(tu,t);
	...
	redo (idx,Rp):= bpm.hasMoreElements(R);
exit (idx,b);
	bpm.close();
@end example

The partitioned BATs are particularly useful during query
optimization. However, it only works if the BAT identifier
can be determined at compile time. For SQL it can be simply looked up in
the catalog as part of the preparatory optimizer step.
The same problem handled by an optimizer produces the plan:
@example
	bpm.open();
	R:= bpm.take("myversion",:bat[:oid,:int]); # get the partition alias
	optimizer.bpm();
	T:= algebra.select(R,0,100);
@end example
is translated into 
@example
	bpm.open();
	Ralias:= bpm.take("myversion",:bat[:oid,:int]);
	R0:= bpm.take(Ralias, 0@0, 100@0, nil:int,nil:int); 
	R1:= bpm.take(Ralias, 101@0, 200@0, nil:int,nil:int); 
	R2:= bpm.take(Ralias, 201@0, nil:oid, nil:int,nil:int); 
	R:= mat.new(R0,R1,R2);
@end example
It is up to the @sc{mat} optimizer to decide about
plan expansion or an iterator approach.

A decision should be made over what happens if you
take out a fragment dat does not align with any
of the boundaries. For now, it generates an exception.


@- Partition updates
The content of the partitions is preferrable updated in bulk.
This calls for accumulation of insertions/deletions in pending
update BATs, as already performed in the SQL code generator.
Once the transaction is commited, the updates are propagated
(in parallel) to all partitions.
@example
	bpm.open();
	R := bpm.take("myversion",:bat[:oid,:int]); # get the partition alias
	bpm.insert(R, Rinsert);	# handle pending inserts
	bpm.delete(R, Rdelete);	# handle pending deletes
	bpm.deposit(R, Rinsert);# handle pending inserts and forget content 
	bpm.close();
@end example

The @sc{bpm.deposit(R,Rinsert)} takes over the content
of the bat Rinsert, leaving an empty BAT behind.
This does not hold for the insert and delete cases.

It remains possible to retrieve a partition and directly
insert elements, but then it is up to the compiler to
ensure that the boundery conditions are met.

@- Partitioned results
In many situations, you would like to keep the partial results
as a partitioned BAT again.
The easiest solution is to remember the partitions as temporary 
variables during optimization, e.g. as properties of all fragment variables.

However, a persistent partitioned BAT can also be created first,
whose partitions are empty. Subsequently, we insert the
temporary results. 
Depending on the fragmentation criteria, pieces may align
with the pieces known, or lead to a redistribution of the
buns to the correct bats.

The previous plan for this becomes
@example
	bpm.open();
	Tmp := bpm.take("tmp",:bat[:oid,:int]);
	bpm.rangePartition(Tmp,nil:oid,100:oid,nil:int,nil:int);
	bpm.rangePartition(Tmp,101:oid,200:oid,nil:int,nil:int);
	bpm.rangePartition(Tmp,201:oid,nil:oid,nil:int,nil:int);

	Ralias:= bpm.take("myversion",:bat[:oid,:int]); # get the partition alias
	R0:= bpm.take(Ralias, 0@0, 100@0, nil:int,nil:int); 
	T0:= algebra.select(R0,0,100);
	bpm.deposit(Tmp,T0);

	R1:= bpm.take(Ralias, 101@0, 200@0, nil:int,nil:int); 
	T1:= algebra.select(R1,0,100);
	bpm.deposit(Tmp,T1);

	R2:= bpm.take(Ralias, 201@0, nil:oid, nil:int,nil:int); 
	T2:= algebra.select(R2,0,100);
	bpm.deposit(Tmp,T2);
@end example
The rationale for this approach is that re-distribution
of temporary results are hidden behind the @sc{bpm} interface.
The only decision that should be taken by the optimizer is
the fragmentation criteria for the temporary results.

For temporary results the range bounds need not be
stored in the bpm. Instead, the mat approach could
be used to reduce the plan size.
@example
	bpm.open();
	Ralias:= bpm.take("myversion",:bat[:oid,:int]); # get the partition alias
	R0:= bpm.take(Ralias, 0@0, 100@0, nil:int,nil:int); 
	T0{hlow=0@0,hhigh=0@0,tlow=nil:int,thigh=nil:int}:=algebra.select(R0,0,100);

	R1:= bpm.take(Ralias, 101@0, 200@0, nil:int,nil:int); 
	T1{hlow=0@0,hhigh=0@0,tlow=nil:int,thigh=nil:int}:=algebra.select(R1,0,100);

	R2:= bpm.take(Ralias, 201@0, nil:oid, nil:int,nil:int); 
	T2{hlow=0@0,hhigh=0@0,tlow=nil:int,thigh=nil:int}:=algebra.select(R2,0,100);
	R:= mat.new(T0,T1,T2);
@end example

@- Partition selection
The select operation can be overloaded in the BPM to
improve processing further. For example, the operation
@example
	t := bpm.select(Ralias,0,100);
@end example
extracts portions of all three partitions and creates 
a non-partitioned result BAT. 
There is no information on the partitions involved in this operation
for the optimizer.


The lifetime of a partitioned table is inherited from its components. 
How to detect that a temporary BAT is removed from the BBP?

@mal
module bpm;

command open():void
address BPMopen
comment "Locate and open the BAT partition box";

command close():void
address BPMclose
comment "Save and close the BAT partition box ";

command destroy():void
address BPMdestroy
comment "Destroy the BAT partition box";

command deposit(nme:str,b:bat[:oid,:any_2]) :void
address BPMdepositName
comment "Create a new partitioned BAT by name";
command deposit(alias:bat[:oid,:any_2],b:bat[:oid,:any_2]) :void
address BPMdeposit
comment "Enter the content of a BAT into a partitioned one.
The side effice is that the argument BAT is empty afterwards.";

@-
The partitioning is handled inside the module.
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
address BPMrange
comment "Create a range partition on a BAT";
command hashPartition(pb:bat[:oid,:any_2], slots:int):void
address BPMhash
comment "Create a hash partition on a BAT";
command rangeHashPartition(pb:bat[:oid,:any_2],
		ll:oid, lh:oid, slots:int):void
address BPMrangeHash
comment "Create a range and hash index partition on a BAT";
command hashRangePartition(pb:bat[:oid,:any_2],
		slots:int, rl:any_2, lh:any_2):void
address BPMhashRange
comment "Create a range and hash index partition on a BAT";
command derivedPartition(pb:bat[:oid,:any_2], src:bat[:oid,:any_3]):void
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]
address BPMtake
comment "Retrieve the alias for a partitioned BAT";

command take(pb:bat[:oid,:any_2],
		ll:oid, lh:oid, slot:int):bat[:oid,:any_2]
address BPMtakeRangeHash
comment "Retrieve a single component of a partitioned BAT 
	by range and hash index";
command take(pb:bat[:oid,:any_2], slot:int, 
	rl:any_2, rh:any_2) :bat[:oid,:any_2]
address BPMtakeHashRange
comment "Retrieve a single component of a partitoined BAT 
	by hash index and range";
command take(pb:bat[:oid,:any_2], ll:oid, lh:oid, 
	rl:any_2, rh:any_2):bat[:oid,:any_2]
address BPMtakeRange
comment "Retrieve a single component of a MAT by range";
command take(pb:bat[:oid,:any_2],hh:int,th:int):bat[:oid,:any_2]
address BPMtakeHash
comment "Retrieve a single component of a MAT by hash indices";

command take(pb:bat[:oid,:any_2],idx:int):bat[:oid,:any_2]
address BPMtakePartition
comment "Retrieve a single component of a MAT by index";

command insert(pb:bat[:oid,:any_2],b:bat[:oid,: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
address BPMdelete
comment "Delete elements from the BAT partitions";
command replace(pb:bat[:oid,:any_2],i:bat[:oid,:any_2],
		d:bat[:oid,:any_2]) :void
address BPMreplace
comment "Replace the content of the BAT partitions";

command getNames():bat[:void,:str]
address BPMgetNames
comment "Retrieve the names of all known partitioned BATs";

command discard(alias:bat[:oid,:any_2]) :void
address BPMdiscard
comment "Release a partitioned BAT from the box";

command newIterator()(:int,:str)
address BPMnewIteratorBase
comment "Create an iterator over the partition box";
command hasMoreElements()(:int,:str)
address BPMhasMoreElementsBase
comment "Locate next element in the partition box";

@-
In most situations we would like to iterator over
the components of a single partitioned BAT. 
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]
address BPMnewIterator
comment "Create an iterator over the BAT partitions.";

command newIterator(grp:bat[:oid,:any_2],first:oid,last:oid)
		:bat[:oid,:any_2]
address BPMnewIteratorRng
comment "Create an iterator over the BAT partitions.";

command newIterator(pb:bat[:oid,:any_2], first:oid,last:oid,
		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]
address BPMhasMoreElements
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]
address BPMhasMoreElementsRng4
comment "Localize the next partition for processing.";

command getDimension(b:bat[:oid,:any_2])(first:oid,last:oid, 
	vlow:any_2, vhgh:any_2)
address BPMgetDimension
comment "Obtain the partition boundary values.";

command dump(alias:bat[:oid,:any_2])
address BPMdumpAlias;
command dump()
address BPMdump;

command prelude()
address BPMprelude;

command epilogue()
address BPMepilogue;

bpm.prelude();
@-
@{
BEWARE, it is not protected against concurrent access yet.

@include ../kernel/kprelude.mx
@+ BAT Partition Manager Implementation
The implementation is organized around a shared box, which should
be saved between session. It is up to other layers to ensure that BATs
being deleted are also removed from the partition box to avoid
mis-represented information.

The internal data structure is used as a cache for improved access.
@h
#ifndef _MAL_BPM
#define _MAL_BPM
#include "mal.h"
#include "mal_client.h"
#include "mal_interpreter.h"

#ifdef WIN32
#ifndef LIBBPM
#define bpm_export extern __declspec(dllimport)
#else
#define bpm_export extern __declspec(dllexport)
#endif
#else
#define bpm_export extern
#endif

bpm_export str BPMopen(void);
bpm_export str BPMclose(int *ret);
bpm_export str BPMdestroy(int *ret);
bpm_export str BPMdepositName(int *ret, str *nme, int *src);
bpm_export str BPMdeposit(int *ret, int *bid, int *src);

bpm_export str BPMrange(int *ret, int *bid, ptr *hl, ptr *hh, ptr *tl, ptr *th);
bpm_export str BPMhash(int *ret, int *bid, int *hslots, int *tslots);
bpm_export str BPMrangeHash(int *ret, int *bid, ptr *hl, ptr *hh, int *tslots);
bpm_export str BPMhashRange(int *ret, int *bid, int *hslots, ptr *tl, ptr *th);
bpm_export str BPMderived(int *ret, int *bid, int *src);

bpm_export str BPMtake(int *ret, str *nme);
bpm_export str BPMtakeRange(int *ret, int *bid, ptr *hl, ptr *hh, ptr *tl, ptr *th);
bpm_export str BPMtakeHash(int *ret, int *bid, int *hslots, int *tslots);
bpm_export str BPMtakeRangeHash(int *ret, int *bid, ptr *hl, ptr *hh, int *tslots);
bpm_export str BPMtakeHashRange(int *ret, int *bid, int *hslots, ptr *tl, ptr *th);

bpm_export str BPMtakePartition(int *ret, int *bid, int *idx);

bpm_export str BPMinsert(int *ret, int *bid, int *ins);
bpm_export str BPMdelete(int *ret, int *bid, int *del);
bpm_export str BPMreplace(int *ret, int *bid, int *ins, int *del);

bpm_export str BPMgetNames(int *bid);
bpm_export str BPMdiscard(int *ret, int *bid);

bpm_export str BPMnewIteratorBase(int *ret, str *grp);
bpm_export str BPMhasMoreElementsBase(int *ret, str *nme);

bpm_export str BPMnewIterator(int *res, int *grp);
bpm_export str BPMnewIteratorRng(int *res, int *grp, ptr *first, ptr *last);
bpm_export str BPMnewIteratorRng4(int *res, int *grp, ptr *first, ptr *last,
	ptr *vlow, ptr *vhgh);

bpm_export str BPMhasMoreElements(int *res, int *grp);
bpm_export str BPMhasMoreElementsRng(int *res, int *grp, ptr *first, ptr *last);
bpm_export str BPMhasMoreElementsRng4(int *res, int *grp, ptr *first, ptr *last,
	ptr *vlow, ptr *vhgh);

bpm_export str BPMgetDimension(ptr *first, ptr *last,
	ptr *vlow, ptr *vhgh, int *bid);
bpm_export str BPMdump(int *ret);
bpm_export str BPMdumpAlias(int *ret, int *bid);
bpm_export str BPMprelude(int *ret);
bpm_export str BPMepilogue(int *ret);
#endif
@-
The partition manager uses its private memory mapped catalog of persistent BATs.
The handle returned is a BAT that represents the whole group.
They are further administered  by type
@c
#include "mal_config.h"
#include "bpm.h"