zbtree.cpp

实现一个精简型单用户SQL引擎(DBMS)MiniSQL

    if(SearchPath[i] == ptr)
      return SearchPath[i-1];  
  }
  throw 1022;  // search path error
}

// get the start key address of the b+ tree
Key_Location ZBTree::GetStart()
{
  ZBTree_Index idx;
  idx.ReadHeadFromFile();
  _F_FileAddr ptr = idx.GetRoot();
  
  for(;;)
  {
    ZBTree_Node Node;
    Node.ReadNodeFromFile(ptr);
    if( 1 == Node.IsLeaf )  // reach the leaf
    {
      Key_Location temp;
      temp.ptr = ptr;
      temp.offset = 0;
      return temp;  // return the first address of the first Key in the first node
    }
    else
      ptr = Node.p[0];
  }
}

// change the current path to *.idx,which means we are using index file now.
// It is only used for buffer to identify the correct file
void ZBTree::SetPath()
{
  char address[256];
  strcpy(address,CurLocation);
  strcat(address,CurRelationName);
  strcat(address,".idx");
  _M_File test = Buffer[address]; 
}
// get the end key address of the b+ tree
Key_Location ZBTree::GetEnd()
{
  ZBTree_Index idx;
  idx.ReadHeadFromFile();
  _F_FileAddr ptr = idx.GetRoot();
  
  for(;;)
  {
    ZBTree_Node Node;
    Node.ReadNodeFromFile(ptr);
    if( 1 == Node.IsLeaf )  // reach the leaf
    {
      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];
  }
}

// get the current file address 
_F_FileAddr ZBTree::GetCurRecAddr(Key_Location KeyLoca)
{
  ZBTree_Node node;
  node.ReadNodeFromFile(KeyLoca.ptr);
  return node.p[KeyLoca.offset];
}
// TODO
// move to the address of the next key
Key_Location ZBTree::MoveToNextKey(Key_Location KeyLoca)
{
  Key_Location temp;
  ZBTree_Node node; 
  node.ReadNodeFromFile(KeyLoca.ptr);

  if( node.MaxItem == KeyLoca.offset )  // p[ItemMax]
  {
    if( 0 == node.p[node.MaxItem].ulFilePageID )  // the end of the tree
      temp.ptr.ulFilePageID = 0;  // 0 is a mark *******
    else
    {
      // get the address of the first key in the next node 
      ZBTree_Node NextNode;
      NextNode.ReadNodeFromFile(node.p[node.MaxItem]);
      temp.ptr = node.p[node.MaxItem];
      temp.offset = 0;
    }    
    return temp;
  }
  int i;
  i = KeyLoca.offset + 1;
  // the next key is empty or p[MaxItem]
  if( 0 == node.p[i].ulFilePageID || node.MaxItem == i)    
  {
    i = node.MaxItem;  // the last p[MaxItem]
  
    if( 0 == node.p[i].ulFilePageID )  // the end of the tree
      temp.ptr.ulFilePageID = 0;  // 0 is a mark *******
    else
    {
      // get the address of the first key in the next node 
      ZBTree_Node NextNode;
      NextNode.ReadNodeFromFile(node.p[i]);
      temp.ptr = node.p[i];
      temp.offset = 0;
    }
  }
  else
  {
    // the address of the next key in the current node 
    temp.ptr = KeyLoca.ptr;
    temp.offset = KeyLoca.offset + 1;
  }
  return temp;
}


// replace the key equal to the second parament with the third parement
// in the node of first parament 
void ZBTree::ReplaceKey(_F_FileAddr ptr,pKey_Attr Key_,pKey_Attr Key)
{
  ZBTree_Node Node;
  Node.ReadNodeFromFile(ptr);
  for(int i=0;i<Node.ItemOnNode;i++)
  {
    if( 0 == Node.Compare(Node.k[i],Key_) )
      break;
  }
  Node.DeleteKey(Node.k[i]);  // delete the space of k[i]
  Node.k[i] = Key;
  Node.WriteNodeToFile(ptr);
}
// set the ancestor or successor according to the parent 
void ZBTree::SetL_(_F_FileAddr L,_F_FileAddr* pL_,pKey_Attr* ppPrimKey_,bool* pIsLLeftOfL_)
{
  ZBTree_Node ParentNode;
  ParentNode.ReadNodeFromFile(Parent(L));
  for(int i=0;i<=ParentNode.ItemOnNode;i++)
  {
    if( ParentNode.p[i] == L )
      break;
  }
  if( 0 == i )  // L is the mostleft 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;
  }
}
// Insert a empty node in the index file,return the address of the node in file
_F_FileAddr ZBTree::CreateNodeInFile()
{
  ZBTree_Index idx;
  idx.ReadHeadFromFile();
  
  if( 0 == idx.IdxHead.FirstEmptyBlock.ulFilePageID ) // the index has no empty block
  {
    // 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)  // the last empty block
      flag = 1;
    _F_FileAddr temp = idx.IdxHead.FirstEmptyBlock;
    if( 1 == flag)  // now,there is no 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 the next block
      ZBTree_Node NextNode;
      NextNode.ReadNodeFromFile(temp);
      idx.IdxHead.FirstEmptyBlock = NextNode.Next_Empty_Block;
    }
    idx.WriteHeadToFile();
    return temp;   
  }  
  
}

// create a new root ----- p0 | pPrimKey | p1
void ZBTree::CreateNewRoot(_F_FileAddr p0,pKey_Attr pPrimKey,_F_FileAddr p1)
{
  _F_FileAddr Newroot;
  Newroot = CreateNodeInFile();  // get a new node
  ZBTree_Index idx;
  idx.ReadHeadFromFile();
  idx.IdxHead.root = Newroot;  // change the index head
  idx.WriteHeadToFile();

  ZBTree_Node Root;
  Root.IsLeaf = 0;  // it can't be a root
  Root.ItemOnNode = 1;
  Root.p[0] = p0;
  Root.p[1] = p1;
  Root.k[0] = pPrimKey;
  Root.WriteNodeToFile(Newroot);
}


// redistribute the tree
void ZBTree::ReDistribute(ZBTree_Node* pNodeL,ZBTree_Node* pNodeL_,pKey_Attr pPrimKey_,bool IsLLeftOfL_,_F_FileAddr L)
{
  int m;
  if( !IsLLeftOfL_ )  // L_ <  L
  {
    if( !pNodeL->IsLeaf ) // it isn't a leaf
    {
      m = pNodeL_->ItemOnNode;
      //  move all k[i] and p[i] 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++;
      ReplaceKey(Parent(L),pPrimKey_,pNodeL_->k[m-1]);  // replace the parent key V'
      pNodeL_->k[m-1] = NULL;
      pNodeL_->p[m].ulFilePageID = 0;
      pNodeL_->p[m].uiOffset     = 0;
      pNodeL_->ItemOnNode--;
    }
    else  // it is a leaf
    {
       m = pNodeL_->ItemOnNode - 1;
       // move 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++;
       pKey_Attr temp = pNodeL_->CreateKey(pNodeL_->k[m]);
       ReplaceKey(Parent(L),pPrimKey_,temp);
       pNodeL->DeleteKey(pPrimKey_);
       pNodeL_->k[m] = NULL;
       pNodeL_->p[m].ulFilePageID = 0;
       pNodeL_->p[m].uiOffset     = 0;
       pNodeL_->ItemOnNode--;
    }
  }
  else  //L < L_
  {
    if( !pNodeL->IsLeaf )  // it isn't a leaf
    {
      pNodeL->k[pNodeL->ItemOnNode] = pPrimKey_;
      pNodeL->p[pNodeL->ItemOnNode + 1] = pNodeL_->p[0];
      pNodeL->ItemOnNode++;
      ReplaceKey(Parent(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  // it is a leaf
    {
      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(Parent(L),pPrimKey_,temp);
      pNodeL->DeleteKey(pPrimKey_);
    }
  }
  
}

void ZBTree::Insert(pKey_Attr pPrimKey,_F_FileAddr pRec)
{
  SetPath();  // set path to *.idx
  Key_Location KeyLoca;
  if( Search(pPrimKey,&KeyLoca) ) // find the primary key
    throw  1024;   // the primary is existed;
  else
    Insert_Entry(KeyLoca.ptr,pPrimKey,pRec);
}


void ZBTree::Insert_Entry(_F_FileAddr L,pKey_Attr pPrimKey,_F_FileAddr pRec)
{
  ZBTree_Node NodeL;
  NodeL.ReadNodeFromFile(L);
  if( NodeL.ItemOnNode < NodeL.MaxItem )  // the node L has space for insert
  {
    if( 1 == NodeL.IsLeaf )  // it is a leaf node
      NodeL.InsertKeyInLeaf(pRec,pPrimKey);
    else                     // it is a non-leaf node
      NodeL.InsertKeyNotInLeaf(pPrimKey,pRec);
  }
  
  else
  {  
    _F_FileAddr L_ = CreateNodeInFile();  // create a node named L_
    ZBTree_Node NodeL_;
    pKey_Attr pPrimKey_;
    
    int n = NodeL.MaxItem + 1;
    if( 1 == NodeL.IsLeaf )  // the node L is a leaf
    {
      NodeL_.IsLeaf = 1;     // the Node L_ is a leaf
      int m; 
      if( !(n%2) )
        m = n/2 ;  // n = NodeL.MaxItem + 1
      else m = n/2 + 1;
      // three condition k[m-1],pPrimKey,k[m]
      if( NodeL.Compare(NodeL.k[m-1],pPrimKey) > 0) // k[m-1] > pPrimKey
      { 
        m--;                                  
        pPrimKey_ = NodeL.CreateKey(NodeL.k[m]);
      }
      else if( (m == n - 1) || ( NodeL.Compare(NodeL.k[m],pPrimKey) > 0 ) ) // k[m] > pPrimKey >= k[m-1]
        pPrimKey_ = NodeL.CreateKey(pPrimKey);
      else
        pPrimKey_ = NodeL.CreateKey(NodeL.k[m]);    // pPrimKey > k[m]

      for(int i=m;i<NodeL.MaxItem;i++)
      {
        NodeL_.p[i-m] = NodeL.p[i];
        NodeL_.k[i-m] = NodeL.k[i];
        NodeL.k[i] = NULL;  // it is useful
        NodeL.p[i].ulFilePageID = 0;
        NodeL.p[i].uiOffset     = 0;
        NodeL_.ItemOnNode++;
        NodeL.ItemOnNode--;
      }
      if(NodeL.Compare(pPrimKey,pPrimKey_) < 0)  // V < V'
        NodeL.InsertKeyInLeaf(pRec,pPrimKey);
      else
        NodeL_.InsertKeyInLeaf(pRec,pPrimKey);
    }
    else  // the node L isn't a leaf
    {
      NodeL_.IsLeaf = 0; // the Node L_ isn't  a leaf
      int m;
      if( !(n%2) )
        m = n/2 ;  // n = NodeL.MaxItem + 1
      else m = n/2 + 1;
      m = NodeL.MaxItem - m;
      
      // three condition k[m],pPrimKey,k[m+1]
      if( NodeL.Compare(pPrimKey,NodeL.k[m+1]) > 0 )   // k[m] < k[m+1] < pPrimKey
      {
        m++;
        pPrimKey_ = NodeL.CreateKey(NodeL.k[m]);
      }
      else if( NodeL.Compare(pPrimKey,NodeL.k[m]) > 0 )// k[m] < pPrimKey <= k[m+1]
      {
        m++;
        pPrimKey_ = NodeL.CreateKey(pPrimKey);
      }
      else                                             // pPrimKey <= k[m] < k[m+1]
      {
        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];
        NodeL.k[i] = NULL;  // it is useful
        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.InsertKeyNotInLeaf(pPrimKey,pRec);
      else 
        NodeL_.InsertKeyNotInLeaf(pPrimKey,pRec);
      NodeL_.DeleteKeyV_();  // delete (nil,V'),the fisrt key in NodeL_
    }
    if( !IsRoot(L) )  // L is not root 
      Insert_Entry(Parent(L),pPrimKey_,L_);
    else
      CreateNewRoot(L,pPrimKey_,L_);  
    if( 1 == NodeL.IsLeaf)  // modify the point of next node 
    {
      NodeL_.p[NodeL.MaxItem] = NodeL.p[NodeL.MaxItem];
      NodeL.p[NodeL.MaxItem]  = L_;
    }
    NodeL_.WriteNodeToFile(L_);
  }

  NodeL.WriteNodeToFile(L);
}

void ZBTree::Delete(pKey_Attr pPrimKey,_F_FileAddr &pRec)
{
  SetPath();  // set path to *.idx
  Key_Location KeyLoca;
  if( Search(pPrimKey,&KeyLoca) ) // find the primary key
  {
    pRec = GetCurRecAddr(KeyLoca);  // return the record address used by Record module
    Delete_Entry(KeyLoca.ptr,pPrimKey,pRec);    
  }
  else
    throw  1023;   // the primary key isn't existed;
}

void ZBTree::Delete_Entry(_F_FileAddr L,pKey_Attr pPrimKey,_F_FileAddr pRec)
{
  ZBTree_Node NodeL;
  NodeL.ReadNodeFromFile(L);

  // delete the (V,P) entry
  if( 1 == NodeL.IsLeaf )
    NodeL.DeleteKeyInLeaf(pPrimKey);
  else 
    NodeL.DeleteKeyNotInLeaf(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 
  }
  // different from book
  else if( !IsRoot(L) && NodeL.IsNotEnoughPt() ) // isn't root and not enough points
  {
    bool IsLLeftOfL_;
    _F_FileAddr L_;
    pKey_Attr pPrimKey_;
    SetL_(L,&L_,&pPrimKey_,&IsLLeftOfL_);  // set the ancestor or successor 
    
    ZBTree_Node NodeL_;
    NodeL_.ReadNodeFromFile(L_);
    if( CanMerge(&NodeL,&NodeL_))  // the nodes can be merged
    {
      if(IsLLeftOfL_) 
        SwapVariables(L,L_); // swap the points in the parent node
      Merge(&NodeL,&NodeL_,pPrimKey_,IsLLeftOfL_);  //**********
      if( NodeL.IsLeaf && IsLLeftOfL_ )  // differe from book 
      {
        SwapVariables(L,L_);
      }        

      Delete_Entry(Parent(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 
    {
      ReDistribute(&NodeL,&NodeL_,pPrimKey_,IsLLeftOfL_,L);
    }
    NodeL_.WriteNodeToFile(L_);
  }
 
  NodeL.WriteNodeToFile(L);  // if though the node is delete
}