gist.cpp

此文件包含了在linux下实现tpr-tree索引的源代码

//--------------------------------------------------------------------
//        GiST.cpp
//        --------
//
//        GiST - Generalized Search Tree 
//        June 2001 release, Aalborg University
//        
//        This file is a revised version of a part of the original 
//        libGiST release (version  0.9beta1)     
//        Copyright (c) 1996, Regents of the University of California
//

#include <string.h>

#include "GiST.h"
#include "GiSTpath.h" // for GiSTRootPage

TruePredicate truePredicate;

//------------------------------------------------
GiST::GiST() :
      store (NULL),
// Observer support
      observer (NULL)
//
{
    isOpen = 0;
    debug = 0;
}

//------------------------------------------------
void
GiST::Create(const char *filename)
{
    if (IsOpen()) return;

    if (!store) store = CreateStore();

    store->Create(filename);
    if (!store->IsOpen())	return;

    store->Allocate();

    GiSTnode *node = NewNode(this);
    node->Path().MakeRoot();

    WriteNode(node);
    delete node;
    isOpen = 1;

// Observer support
    if (observer) observer->Inform(ON_CREATE);
//
}

//------------------------------------------------
void
GiST::Open(const char *filename)
{
    if (IsOpen())	return;

    if (!store) store = CreateStore();

    store->Open(filename);
    if (!store->IsOpen())	return;

    isOpen = 1;

// Observer support
    if (observer) observer->Inform(ON_OPEN);
//
}

//------------------------------------------------
void
GiST::Close()
{
    if (IsOpen())
    {
	    store->Close();
	    isOpen = 0;
    }
// Observer support
    if (observer) observer->Inform(ON_CLOSE);
//
}

//------------------------------------------------
void
GiST::Insert(const GiSTentry &entry)
{
   GiSTnode* leaf = InsertHelper(entry, 0);

   CorrectTree(leaf);
   delete leaf;
}

//------------------------------------------------
GiSTnode*
GiST::InsertHelper(const GiSTentry &entry, int level, GiSTnode* root)
{
    GiSTnode *leaf;

// Observer support
    if (observer) observer->Inform(ON_INSERTSTART);
//
    leaf = ChooseSubtree(GiSTRootPage, entry, level, root);
    if (!leaf->NumEntries()) leaf->SetLevel(level);
    leaf->Insert(entry);

// Observer support
    if (observer) observer->Inform(ON_INSERTEND);
//
    return leaf;
}

//------------------------------------------------
static int OrphanLevel (GiSTlist<GiSTentry*>* listarray)
{
   for (int i = GIST_MAX_LEVELS - 1; i >= 0; i--)
      if (!listarray[i].IsEmpty()) return i;
   
   return -1;
}

//------------------------------------------------
void 
GiST::CorrectTree  (GiSTnode* node)
{
   bool                 splitvec[GIST_MAX_LEVELS];
   GiSTlist<GiSTentry*> orphans[GIST_MAX_LEVELS];
   int                  level;
   GiSTentry*           entry;
   GiSTnode*            tnode;
   GiSTnode*            root = NULL;  
 
   for (level = 0; level < GIST_MAX_LEVELS; level++) splitvec[level] = false;
      
   PropogateUp(node, splitvec, orphans, root);
   for (level = OrphanLevel(orphans); level != -1; level = OrphanLevel(orphans))
   {
      entry = orphans[level].RemoveFront();
      tnode  = InsertHelper(*entry, level, root);
      PropogateUp(tnode, splitvec, orphans, root);
      if (tnode != root) delete tnode;     
      delete entry;
   }   
   
   if (root)
   {
      if (!root->IsLeaf() && root->NumEntries() == 1) ShortenTree(root);
      if (root != node) delete root;
   }
}

//------------------------------------------------
void      
GiST::PropogateUp (GiSTnode* node, bool* splitvec, 
                   GiSTlist<GiSTentry*>* orphans, GiSTnode*& root)
{
   if (!node->Path().IsRoot() && node->IsUnderFull(*store))
   {
      CondenseTree(node, splitvec, orphans, root);
      return;
   }
   if (node->IsOverFull(*store)) 
   {
      if (ForcedReinsert() && !node->Path().IsRoot() && !splitvec[node->Level()])
         OverflowTreatment(node, splitvec, orphans, root);
      else Split(node, splitvec, orphans, root);
      return;
   }   
   WriteNode(node);

   AdjustKeys(node, splitvec, orphans, root);     
}

//------------------------------------------------
void
GiST::OverflowTreatment(GiSTnode *node, bool* splitvec, 
                        GiSTlist<GiSTentry*>* orphans, GiSTnode*& root)
{
   GiSTlist<GiSTentry*> deleted;

// Observer support
   if (observer) observer->Inform(ON_OVERFLOWTSTART, *node);
//
   // remove the "top" p entries from the node
   deleted = RemoveTop(node);

// Observer support
   if (observer) observer->Inform(ON_REMOVETOP, deleted);
//
   // note that we've seen this level already
   splitvec[node->Level()] = true;
    
   // Append deleted entries to orphans list
   while (!deleted.IsEmpty())
      orphans[node->Level()].Append(deleted.RemoveFront());

// Observer support
   if (observer) observer->Inform(ON_OVERFLOWTEND);
//
   return PropogateUp(node, splitvec, orphans, root);  
}

//------------------------------------------------
GiSTlist<GiSTentry*>
GiST::RemoveTop(GiSTnode *node)
{
    GiSTlist<GiSTentry*> deleted;
    int count = node->NumEntries();

    // default: remove the first ones on the page
    int num_rem = (int)((count + 1)*RemoveRatio() + 0.5);
    for (int i = num_rem - 1; i >= 0; i--)
    {
        deleted.Append((GiSTentry *)(*node)[i].Ptr()->Copy());
        node->DeleteEntry(i);
    }
    return(deleted);
}

//------------------------------------------------
void
GiST::AdjustKeys(GiSTnode *node, bool* splitvec, 
                 GiSTlist<GiSTentry*>* orphans, GiSTnode*& root)
{
    GiSTnode *P;
    GiSTpath  parent_path;

    if (node->Path().IsRoot()) 
    {
       root = node;
       return;
    }

    // Read in node's parent
    parent_path = node->Path();
    parent_path.MakeParent();
    P = parent_path.IsRoot() && root ? root : ReadNode(parent_path);

    // Get the old entry pointing to node
    GiSTentry *entry = P->SearchPtr(node->Path().Page());
    assert(entry != NULL);

    // Get union of the node
    GiSTentry *actual = node->Union();
    actual->SetPtr(node->Path().Page());

    if (!entry->IsEqual(*actual))
    {
       int pos = entry->Position();
       P->DeleteEntry(pos);
       P->InsertBefore(*actual, pos);
       PropogateUp(P, splitvec, orphans, root);
    }

    if (P != root) delete P;
    delete actual;
    delete entry;
}

//------------------------------------------------
void
GiST::Sync()
{
    store->Sync();
}

//------------------------------------------------
//  Delete
//  Maximum one object is deleted (first satisfying the predicate)
//
 
int
GiST::Delete(const GiSTpredicate& pred)
{
    int num = 0;
    GiSTentry*  e;
    GiSTcursor* c;

// Observer support
    if (observer) observer->Inform(ON_DELETESTART);
//
    c = Search(pred);

    if ((e = c->Next()) != NULL)
    {
      GiSTpath path = c->Path();

      // Read in the node that this belongs to
      GiSTnode *node = ReadNode(path);
      node->DeleteEntry(e->Position());

      CorrectTree (node);
      delete node;
      delete e;

      num++;
    }
    delete c;

// Observer support
    if (observer) observer->Inform(ON_DELETEEND);
//
    return num;
}

//------------------------------------------------
void
GiST::ShortenTree(GiSTnode* root)
{
    GiSTpath path;

    path.MakeRoot();
    path.MakeChild((*root)[0]->Ptr());
    GiSTnode *child = ReadNode(path);
    DeallocateNode(child);

// Observer support
    if (observer) observer->Inform(ON_DEALOCATE, *child);
//
    child->SetSibling(0);
    child->Path().MakeRoot();
    WriteNode(child);

// Observer support
    if (observer) observer->Inform(ON_SHORTENTREE, *child);
//
    delete child;
}

//------------------------------------------------
// handle underfull leaf nodes
void
GiST::CondenseTree(GiSTnode *node, bool* splitvec, 
                   GiSTlist<GiSTentry*>* orphans, GiSTnode*& root)
{
// Observer support
   if (observer) observer->Inform(ON_CONDENSETREESTART);
//
   GiSTpath parent_path;
   parent_path = node->Path();
   parent_path.MakeParent();

   GiSTnode  *P  = parent_path.IsRoot() && root ? root : ReadNode(parent_path);
   GiSTentry *En = P->SearchPtr(node->Path().Page());

   assert(En != NULL);
   assert(!IsOrdered());

   GiSTlist<GiSTentry*> list = node->Search(truePredicate);
   while (!list.IsEmpty())
      orphans[node->Level()].Append(list.RemoveFront());

   node->DeleteAll();    
   DeallocateNode(node);

   P->DeleteEntry(En->Position());
    
   PropogateUp(P, splitvec, orphans, root);

   delete En;
   if (P != root) delete P;

// Observer support
   if (observer) observer->Inform(ON_CONDENSETREEEND);
//
}

//------------------------------------------------
GiSTnode*
GiST::ChooseSubtree(GiSTpage page, const GiSTentry &entry, 
                    int level, GiSTnode* root)
{
    GiSTnode *node;
    GiSTpath path;

// Observer support
    if (observer) observer->Inform(ON_CHOOSESUBTREESTART);
//
    for (;;)
    {
       path.MakeChild(page);
       node = page == GiSTRootPage && root ? root : ReadNode(path);
       if (level == node->Level() || node->IsLeaf() || !node->NumEntries()) break;
       page = node->SearchMinPenalty(entry);

// Observer support
/*
       if (observer)
       {
          GiSTentry *e = node->SearchPtr(page);
          observer->Inform(ON_SEARCHMINPENALTY, *e);
          // the node level is one less than that of the current node
          if (node->Level() == 1) 
             observer->Inform(ON_CHOOSENODE, entry, *e, page);   
          delete e;
       }
*/
//
       if (node != root) delete node;
    }

// Observer support
    if (observer) observer->Inform(ON_CHOOSESUBTREEEND);
//
    return node;
}

//------------------------------------------------
void
GiST::Split(GiSTnode *node, bool* splitvec, 
				GiSTlist<GiSTentry*>* orphans, GiSTnode*& root)
{
    bool new_root = false;

    if (node->Path().IsRoot())
    {
  	    new_root = true;
	    node->Path().MakeChild(store->Allocate());
    }

    GiSTnode *node2 = node->PickSplit();
    node2->Path().MakeSibling(store->Allocate());
   
    node2->SetSibling(node->Sibling());                     
    node->SetSibling(node2->Path().Page());

    WriteNode(node);
    WriteNode(node2);

// Observer support
    if (observer) observer->Inform(ON_SPLIT, *node, *node2);
//

    GiSTentry *e1 = node->Union();
    GiSTentry *e2 = node2->Union();
    e1->SetPtr(node->Path().Page());
    e2->SetPtr(node2->Path().Page());

    // Create new root if root is being split
    if (new_root)
    {
       root = NewNode(this);
       root->SetLevel(node->Level() + 1);
       root->InsertBefore(*e1, 0);
       root->InsertBefore(*e2, 1);
       root->Path().MakeRoot();
       WriteNode(root);

// Observer support
       if (observer) observer->Inform(ON_NEWROOT, *root);
//
    }
    else
    {
       // Insert entry for N' in parent
       GiSTpath parent_path = node->Path();
       parent_path.MakeParent();

       GiSTnode *parent = parent_path.IsRoot() && root ? root : ReadNode(parent_path);

       // Find the entry for N in parent
       GiSTentry *e = parent->SearchPtr(node->Path().Page());
       assert(e != NULL);

       // Insert the new entry right after it
       int pos = e->Position();
       parent->DeleteEntry(pos);
       parent->InsertBefore(*e1, pos);
       parent->InsertBefore(*e2, pos+1);
       delete e;

       PropogateUp(parent, splitvec, orphans, root);

       if (parent != root) delete parent;
    }
    delete node2;
    delete e1;
    delete e2;
}

//------------------------------------------------
GiSTnode*
GiST::ReadNode(const GiSTpath& path) const
{
    char *buf = new char[store->PageSize()];

    GiSTnode *node = NewNode((GiST *)this);

    node->Path() = path;

    store->Read(path.Page(), buf);
    node->Unpack(buf);

#ifdef PRINTING_OBJECTS
    if (debug)
    {
       cout << "READ PAGE " << path.Page() << ":\n";
       node->Print(cout);
    }
#endif
    delete buf;

    return node;
}

//------------------------------------------------
void      
GiST::DeallocateNode(GiSTnode *node)
{ 
   node->Commit();
   store->Deallocate(node->Path().Page()); 

// Observer support
   if (observer) observer->Inform(ON_DEALOCATE, *node);
//
}

//------------------------------------------------
void
GiST::WriteNode(GiSTnode *node)
{
   char *buf = new char[store->PageSize()];

// Observer support
   if (observer) observer->Inform(ON_WRITENODE, *node);
//
   // make Purify happy
   memset(buf, 0, store->PageSize());

#ifdef PRINTING_OBJECTS
   if (debug)
   {
      cout << "WRITE PAGE " << node->Path().Page() << ":\n";
      node->Print(cout);
   }
#endif

   node->Pack(buf);
   node->Commit();
   store->Write(node->Path().Page(), buf);
   delete buf;
}

#ifdef PRINTING_OBJECTS

//------------------------------------------------
void
GiST::DumpNode(ostream& os, GiSTpath path) const
{
#ifdef PRINTING_OBJECTS
    GiSTnode *node = ReadNode(path);


    node->Print(os);
    if (!node->IsLeaf())
    {
       GiSTlist<GiSTentry*> list = node->Search(truePredicate);
       while (!list.IsEmpty())
       {
           GiSTentry *e = list.RemoveFront();
           path.MakeChild(e->Ptr());
           DumpNode(os, path);
           path.MakeParent();
           delete e;
       }
    }
    delete node;
#endif
}

//------------------------------------------------
void
GiST::Print(ostream& os) const
{
    GiSTpath path;

    path.MakeRoot();
    DumpNode(os, path);
}

#endif

//------------------------------------------------
GiSTcursor*
GiST::Search(const GiSTpredicate &query) const
{
// Observer support
    if (observer) observer->Inform(ON_SEARCH, query);
//
    return new GiSTcursor(*this, query);
}

//------------------------------------------------
GiSTcursor*
GiST::DebugSearch(const GiSTpredicate &query) const
{
// Observer support
    if (observer) observer->Inform(ON_SEARCH, query);
//
    return new GiSTdebugCursor(*this, query);
}

//------------------------------------------------
GiST::~GiST()
{
    Close();
    if(store) delete store;
}