bptree.cpp

用c语言实现的一个小型dbms

///////////////////////////////////////////////////////////////////////////////////////////////////
//this method returns the address specified by Key_Location
//--
_F_FileAddr BPTree::getCurRecAddr(Key_Location KeyLoca) {
	BPTreeNode node;
	node.readNodeFromFile(KeyLoca.ptr);
	return node.p[KeyLoca.offset];
}

///////////////////////////////////////////////////////////////////////////////////////////////////
//--
_F_FileAddr BPTree::createNodeInFile() {
	BPTree_Index idx;
	idx.readHeadFromFile();
	//the index file has no empty block
	if( 0 == idx.IdxHead.FirstEmptyBlock.ulFilePageID ) {
		// firstly,we should check whether the block overflow the current page
		// If so,use the next page 
		void* CheckOverflow = (void*) new char[BTreeNodeSize];
		_F_FileAddr Next = MemWrite(CheckOverflow, BTreeNodeSize, &idx.IdxHead.FirstNewBlock);
		_F_FileAddr temp = idx.IdxHead.FirstNewBlock;
		idx.IdxHead.FirstNewBlock = Next;
		idx.writeHeadToFile();
		return temp;
	}
	else {
		int flag = 0;  
		if(idx.IdxHead.FirstEmptyBlock ==  idx.IdxHead.LastEmptyBlock)
			flag = 1;
		_F_FileAddr temp = idx.IdxHead.FirstEmptyBlock;
		if( 1 == flag) { 
			//there is just one empty block
			idx.IdxHead.FirstEmptyBlock.ulFilePageID = 0;
			idx.IdxHead.FirstEmptyBlock.uiOffset     = 0;
			idx.IdxHead.LastEmptyBlock.ulFilePageID  = 0;
			idx.IdxHead.LastEmptyBlock.uiOffset      = 0;
		}
		else {
			// change FisrtEmptyBlock point to the next block
			BPTreeNode NextNode;
			NextNode.readNodeFromFile(temp);
			idx.IdxHead.FirstEmptyBlock = NextNode.Next_Empty_Block;
		}
		idx.writeHeadToFile();
		return temp;
	}
}

///////////////////////////////////////////////////////////////////////////////////////////////////
//this method returns the address of the last key in the rightest node
//--
Key_Location BPTree::getEnd() {
	BPTree_Index idx;
	idx.readHeadFromFile();
	_F_FileAddr ptr = idx.getRoot();

	for( ; ; ) {
		BPTreeNode Node;
		Node.readNodeFromFile(ptr);
		if( 1 == Node.IsLeaf ) {
			Key_Location temp;
			temp.ptr = ptr;
			temp.offset = Node.ItemOnNode - 1;
			return temp;  // return the first address of the first Key in the first node
		}
		else
			ptr = Node.p[Node.ItemOnNode];
	}
}

///////////////////////////////////////////////////////////////////////////////////////////////////
//--
void BPTree::insert(pKey_Attr pPrimKey, _F_FileAddr pRec) {
	setPath();
	Key_Location KeyLoca;
	if(search(pPrimKey, &KeyLoca))
		throw 1024;
	else 
		insert_entry(KeyLoca.ptr, pPrimKey, pRec);
}

///////////////////////////////////////////////////////////////////////////////////////////////////
//--
void BPTree::insert_entry(_F_FileAddr L, pKey_Attr pPrimKey, _F_FileAddr pRec) {
	BPTreeNode NodeL;
	NodeL.readNodeFromFile(L);
	//if L has less than n-1 key values
	if(NodeL.ItemOnNode < NodeL.MaxItem) {
		if(NodeL.IsLeaf ==1)
			NodeL.insertKeyInLeaf(pRec, pPrimKey);
		else
			NodeL.insertKeyNotLeaf(pPrimKey, pRec);
	}
	else {
		//create a new node
		_F_FileAddr L_ = createNodeInFile();
		BPTreeNode NodeL_;
		pKey_Attr pPrimKey_;
		int n = NodeL.MaxItem + 1;

		if(NodeL.IsLeaf ==1) {
			NodeL_.IsLeaf = 1;
			int m;
			if( !(n%2) )
				m = n/2 ;
			else 
				m = n/2 + 1;
			//p[0]k[0]...k[m-1]p[m] | k[m]...
			// pPrimKey < k[m-1]
			if( NodeL.compare(NodeL.k[m-1], pPrimKey) > 0) { 
				m--;                                  
				pPrimKey_ = NodeL.createKey(NodeL.k[m]);
			}
			// k[m] > pPrimKey >= k[m-1]
			else if( (m == n - 1) || ( NodeL.compare(NodeL.k[m], pPrimKey) > 0 ) ) 
				pPrimKey_ = NodeL.createKey(pPrimKey);
			else
				// pPrimKey > k[m]
				pPrimKey_ = NodeL.createKey(NodeL.k[m]);
			//copy p[m]....to L_
			for(int i=m; i<NodeL.MaxItem; i++) {
				NodeL_.p[i-m] = NodeL.p[i];
				NodeL_.k[i-m] = NodeL.k[i];
				//erase
				NodeL.k[i] = NULL;
				NodeL.p[i].ulFilePageID = 0;
				NodeL.p[i].uiOffset     = 0;
				NodeL_.ItemOnNode++;
				NodeL.ItemOnNode--;
			}
			if(NodeL.compare(pPrimKey, pPrimKey_) < 0)
				NodeL.insertKeyInLeaf(pRec,pPrimKey);
			else
				NodeL_.insertKeyInLeaf(pRec,pPrimKey);
		}
		else { //NodeL is not a leaf
			NodeL_.IsLeaf = 0;
			int m;
			if( !(n%2) )
				m = n/2 ;
			else 
				m = n/2 + 1;
			m = NodeL.MaxItem - m;
			//p[0],k[0]....k[m]p[m+1] | k[m+1]...
			//pPrimKey > k[m+1]
			if( NodeL.compare(pPrimKey, NodeL.k[m+1]) > 0 ) {
				m++;
				pPrimKey_ = NodeL.createKey(NodeL.k[m]);
			}
			// k[m] < pPrimKey <= k[m+1]
			else if( NodeL.compare(pPrimKey, NodeL.k[m]) > 0 ) {
				m++;
				pPrimKey_ = NodeL.createKey(pPrimKey);
			}
			// pPrimKey <= k[m]
			else {
				pPrimKey_ = NodeL.createKey(NodeL.k[m]);
			}
			// move and delete
			NodeL_.p[0].ulFilePageID = 0;
			NodeL_.p[0].uiOffset     = 0;  
			NodeL_.k[0] = NodeL.k[m];
			NodeL.k[m] = NULL;
			NodeL_.ItemOnNode++;
			NodeL.ItemOnNode--;
			for(int i=m+1; i<NodeL.MaxItem; i++) {
				NodeL_.p[i-m] = NodeL.p[i];
				NodeL_.k[i-m] = NodeL.k[i];
				//erase
				NodeL.k[i] = NULL;
				NodeL.p[i].ulFilePageID = 0;
				NodeL.p[i].uiOffset     = 0;
				NodeL_.ItemOnNode++;
				NodeL.ItemOnNode--;
			}
			NodeL_.p[i-m] = NodeL.p[i];
			NodeL.p[i].ulFilePageID = 0;
			NodeL.p[i].uiOffset     = 0;
			if( NodeL.compare(pPrimKey, pPrimKey_) < 0 )
				NodeL.insertKeyNotLeaf(pPrimKey, pRec);
			else 
				NodeL_.insertKeyNotLeaf(pPrimKey,pRec);
			NodeL_.deleteFirstKey();
		}
		if( !isRoot(L) )
			insert_entry(getParent(L), pPrimKey_, L_);
		else
			createNewRoot(L, pPrimKey_, L_);  
		if(NodeL.IsLeaf == 1) {
			NodeL_.p[NodeL.MaxItem] = NodeL.p[NodeL.MaxItem];
			NodeL.p[NodeL.MaxItem]  = L_;
		}
		NodeL_.writeNodeToFile(L_);
	}
	NodeL.writeNodeToFile(L);
}

//////////////////////////////////////////////////////////////////////////////////////////////////////////

bool BPTree::canCoalesce(BPTreeNode* pNode, BPTreeNode* pNode_) {
	if(pNode->IsLeaf == 1)
		return (pNode->ItemOnNode + pNode_->ItemOnNode <= pNode->MaxItem);
	else
		return (pNode->ItemOnNode + pNode_->ItemOnNode < pNode->MaxItem);   
}

//////////////////////////////////////////////////////////////////////////////////////////////////////////


void BPTree::coalesce(BPTreeNode* pNodeL, BPTreeNode* pNodeL_, pKey_Attr pPrimKey_, bool IsLLeftOfL_) {
	if( 0 == pNodeL->IsLeaf ) {
		if(IsLLeftOfL_) {
			//move right
			int j = pNodeL->ItemOnNode + 1;
			pNodeL_->p[pNodeL_->ItemOnNode + j] = pNodeL_->p[pNodeL_->ItemOnNode];
			for(int i=pNodeL_->ItemOnNode-1; i>=0; i--) {
				pNodeL_->k[i + j] = pNodeL_->k[i];
				pNodeL_->p[i + j] = pNodeL_->p[i];
			}
			//assign V' to the pNodeL_
			pNodeL_->k[j-1] = pPrimKey_;
			pNodeL_->p[j-1] = pNodeL->p[pNodeL->ItemOnNode];
			// move all elements in pNodeL to pNodeL_
			for(i=0; i<pNodeL->ItemOnNode; i++) {
				pNodeL_->p[i] = pNodeL->p[i];
				pNodeL_->k[i] = pNodeL->k[i];
			}
			// set ItemOnNode in pNodeL_
			pNodeL_->ItemOnNode = pNodeL_->ItemOnNode + pNodeL->ItemOnNode + 1;
			pNodeL->initialize();
		}
		else { //L is right of L_
			int j = pNodeL_->ItemOnNode + 1;
			pNodeL_->k[pNodeL_->ItemOnNode] = pPrimKey_; 
			for(int i=0;i<pNodeL->ItemOnNode;i++) {
				pNodeL_->p[i+j] = pNodeL->p[i];
				pNodeL_->k[i+j] = pNodeL->k[i];
			}
			pNodeL_->p[i+j] = pNodeL->p[i];
			pNodeL_->ItemOnNode = pNodeL_->ItemOnNode + pNodeL->ItemOnNode + 1;
			pNodeL->initialize();
		}
	}
	else { // (pNodeL->IsLeaf == 1)
		if(IsLLeftOfL_) {
			int j = pNodeL->ItemOnNode;
			for(int i=0; i<pNodeL_->ItemOnNode; i++) {
				pNodeL->p[i+j] = pNodeL_->p[i];
				pNodeL->k[i+j] = pNodeL_->k[i];
			}
			pNodeL->ItemOnNode += pNodeL_->ItemOnNode;
			pNodeL->p[pNodeL->MaxItem] = pNodeL_->p[pNodeL_->MaxItem];
			pNodeL_->initialize();
		}
		else { // L is right of L_
			int j = pNodeL_->ItemOnNode;
			for(int i=0; i<pNodeL->ItemOnNode; i++) {
				pNodeL_->p[i+j] = pNodeL->p[i];
				pNodeL_->k[i+j] = pNodeL->k[i];
			}
			pNodeL_->ItemOnNode += pNodeL->ItemOnNode;
			pNodeL_->p[pNodeL_->MaxItem] = pNodeL->p[pNodeL->MaxItem];
			pNodeL->initialize();
		}
	}
}

///////////////////////////////////////////////////////////////////////////////////////////////////////////
//this method swaps the pointers to node L and L_

void BPTree::swapVariables(_F_FileAddr L, _F_FileAddr L_) {
	BPTreeNode ParentNode;
	ParentNode.readNodeFromFile(getParent(L));
	int m,n;
	// find the indexes of L and L_ 
	for(int i=0; i <= ParentNode.ItemOnNode; i++) {
		if( ParentNode.p[i] == L )
			m = i;
		if( ParentNode.p[i] == L_ )
			n = i;
	}
	// swap
	_F_FileAddr temp;
	temp = ParentNode.p[m];
	ParentNode.p[m] = ParentNode.p[n];
	ParentNode.p[n] = temp;
	ParentNode.writeNodeToFile(getParent(L));
}

///////////////////////////////////////////////////////////////////////////////////////////////////////////
//this method makes pL_ and ppPrimKey points the infomation that is a neighbour of L
// and pIsLLeftOnL_ indicate whether L is left neighbour of L_ or not

void BPTree::setNb(_F_FileAddr L, _F_FileAddr* pL_, pKey_Attr* ppPrimKey_, bool* pIsLLeftOfL_) {
	BPTreeNode ParentNode;
	ParentNode.readNodeFromFile(getParent(L));
	for(int i=0; i<=ParentNode.ItemOnNode; i++) {
		if( ParentNode.p[i] == L )
			break;
	}
	if( 0 == i ) { // L is the leftest son of the parent node
		(*pL_)          = ParentNode.p[1];
		(*ppPrimKey_)   = ParentNode.createKey(ParentNode.k[0]);
		(*pIsLLeftOfL_) = true;  // L is on the left of L_
	}
	else {
		(*pL_)          = ParentNode.p[i-1];
		(*ppPrimKey_)   = ParentNode.createKey(ParentNode.k[i-1]);
		(*pIsLLeftOfL_) = false;
	}
}

///////////////////////////////////////////////////////////////////////////////////////////////////////////
//this method replaces the key equal to Key_ with Key in the node pointed by ptr  
//--
void BPTree::replaceKey(_F_FileAddr ptr, pKey_Attr Key_, pKey_Attr Key) {
	BPTreeNode Node;
	Node.readNodeFromFile(ptr);
	for(int i=0; i<Node.ItemOnNode; i++) {
		if( 0 == Node.compare(Node.k[i], Key_) )
			break;
	}
	Node.deleteKeySpace(Node.k[i]);
	Node.k[i] = Key;
	Node.writeNodeToFile(ptr);
}

///////////////////////////////////////////////////////////////////////////////////////////////////////////

void BPTree::reDistribute(BPTreeNode* pNodeL, BPTreeNode* pNodeL_, pKey_Attr pPrimKey_, 
						  bool IsLLeftOfL_, _F_FileAddr L) {
	int m;
	if( !IsLLeftOfL_ ) {
		if( !pNodeL->IsLeaf ) {
			m = pNodeL_->ItemOnNode;
			//  move all k[i] and p[i] in L to right by one
			pNodeL->p[pNodeL->ItemOnNode+1] = pNodeL->p[pNodeL->ItemOnNode];
			for(int i=pNodeL->ItemOnNode-1; i>=0; i--) {
				pNodeL->k[i+1] = pNodeL->k[i];
				pNodeL->p[i+1] = pNodeL->p[i];
			}
			pNodeL->k[0] = pPrimKey_;
			pNodeL->p[0] = pNodeL_->p[m];
			pNodeL->ItemOnNode++;
			//replace K_ in parent of L by k[m-1]
			replaceKey(getParent(L), pPrimKey_, pNodeL_->k[m-1]);
			//remove k[m-1] and p[m] from L_
			pNodeL_->k[m-1] = NULL;
			pNodeL_->p[m].ulFilePageID = 0;
			pNodeL_->p[m].uiOffset     = 0;
			pNodeL_->ItemOnNode--;
		}
		else { //(pNodeL->IsLeaf == 1)
			m = pNodeL_->ItemOnNode - 1;
			//  move all k[i] and p[i] in L to right by one
			for(int i=pNodeL->ItemOnNode-1; i>=0; i--) {
				pNodeL->k[i+1] = pNodeL->k[i];
				pNodeL->p[i+1] = pNodeL->p[i];
			}
			pNodeL->k[0] = pNodeL_->k[m];
			pNodeL->p[0] = pNodeL_->p[m];
			pNodeL->ItemOnNode++;
			//replace K_ in parent of L by k[m]
			pKey_Attr temp = pNodeL_->createKey(pNodeL_->k[m]);
			replaceKey(getParent(L), pPrimKey_, temp);
			pNodeL->deleteKeySpace(pPrimKey_);
			//remove p[m] and k[m] from L_ 
			pNodeL_->k[m] = NULL;
			pNodeL_->p[m].ulFilePageID = 0;
			pNodeL_->p[m].uiOffset     = 0;
			pNodeL_->ItemOnNode--;
		}
	}
	else { //(IsLLeftOfL_)
		if( !pNodeL->IsLeaf ) {
			pNodeL->k[pNodeL->ItemOnNode] = pPrimKey_;
			pNodeL->p[pNodeL->ItemOnNode + 1] = pNodeL_->p[0];
			pNodeL->ItemOnNode++;
			replaceKey(getParent(L), pPrimKey_, pNodeL_->k[0]);
			// move pNodeL_ to the left by one
			for(int i=1; i<pNodeL_->ItemOnNode; i++) {
				pNodeL_->k[i-1] = pNodeL_->k[i];
				pNodeL_->p[i-1] = pNodeL_->p[i];
			}
			pNodeL_->p[i-1] = pNodeL_->p[i];

			pNodeL_->k[pNodeL_->ItemOnNode-1] = NULL;
			pNodeL_->p[pNodeL_->ItemOnNode].ulFilePageID = 0;
			pNodeL_->p[pNodeL_->ItemOnNode].uiOffset     = 0;
			pNodeL_->ItemOnNode--;
		}
		else { //(pNodeL->IsLeaf)
			pNodeL->k[pNodeL->ItemOnNode] = pNodeL_->k[0];
			pNodeL->p[pNodeL->ItemOnNode] = pNodeL_->p[0];
			pNodeL->ItemOnNode++;
			for(int i=1; i<pNodeL_->ItemOnNode; i++) {
				pNodeL_->k[i-1] = pNodeL_->k[i];
				pNodeL_->p[i-1] = pNodeL_->p[i];
			}
			pNodeL_->k[pNodeL_->ItemOnNode-1] = NULL;
			pNodeL_->p[pNodeL_->ItemOnNode-1].ulFilePageID = 0;
			pNodeL_->p[pNodeL_->ItemOnNode-1].uiOffset     = 0;
			pNodeL_->ItemOnNode--;
			pKey_Attr temp = pNodeL_->createKey(pNodeL_->k[0]); 
			replaceKey(getParent(L), pPrimKey_, temp);
			pNodeL->deleteKeySpace(pPrimKey_);
		}
	}
}

///////////////////////////////////////////////////////////////////////////////////////////////////////////

void BPTree::myDelete(pKey_Attr pPrimKey, _F_FileAddr& pRec) {
	setPath();
	Key_Location KeyLoca;
	if(search(pPrimKey, &KeyLoca)) {
		pRec = getCurRecAddr(KeyLoca);
		delete_entry(KeyLoca.ptr, pPrimKey, pRec);    
	}
	else
		throw 1023;
}

///////////////////////////////////////////////////////////////////////////////////////////////////////////

void BPTree::delete_entry(_F_FileAddr L, pKey_Attr pPrimKey, _F_FileAddr pRec) {
	BPTreeNode NodeL;
	NodeL.readNodeFromFile(L);
	// delete the (V,P) entry
	if( 1 == NodeL.IsLeaf )
		NodeL.deleteKeyInLeaf(pPrimKey);
	else 
		NodeL.deleteKeyNotLeaf(pPrimKey);
	// L is root and has only one son and not a leaf
	if( isRoot(L) && ( 0 == NodeL.ItemOnNode ) && (0 == NodeL.IsLeaf) )  {
		grantRoot(NodeL.p[0]); // grant root to the son
		deleteNodeInFile(L);   // delete node 
	}
	else 
		if( !isRoot(L) && NodeL.isNotEnoughPoints() )  {// isn't root and not enough points
			//set L_ neighbour of L
			bool IsLLeftOfL_;
			_F_FileAddr L_;
			pKey_Attr pPrimKey_;
			setNb(L, &L_, &pPrimKey_, &IsLLeftOfL_);
    
			BPTreeNode NodeL_;
			NodeL_.readNodeFromFile(L_);
			if( canCoalesce(&NodeL, &NodeL_)) {
				if(IsLLeftOfL_) 
					swapVariables(L,L_); 
				coalesce(&NodeL, &NodeL_, pPrimKey_, IsLLeftOfL_);
				if( NodeL.IsLeaf && IsLLeftOfL_ ) // differe from book 
					swapVariables(L,L_);       

				delete_entry(getParent(L), pPrimKey_, L);
				if( NodeL.IsLeaf && IsLLeftOfL_ )  // L is a leaf and L < L_. Different from the book
					deleteNodeInFile(L_);  // delete the node pointed by L_
				else
					deleteNodeInFile(L);   // delete the node pointed by L
			}
			else 
			  reDistribute(&NodeL, &NodeL_, pPrimKey_, IsLLeftOfL_, L);

			NodeL_.writeNodeToFile(L_);
		}
		NodeL.writeNodeToFile(L);  // if though the node is delete
}

///////////////////////////////////////////////////////////////////////////////////////////////////////////