unit BPlusC;
//*******************************************************
// B+tree class.
//*******************************************************
// Copyright (C) 1998 John Wiley & Sons, Inc.
// All rights reserved. See additional copyright
// information in Readme.txt.
//*******************************************************

interface

const
    // We use FirstName,LastName as the bucket keys.
    // These plus the comma occupy 41 bytes. Adding in
    // the Longint child pointers, we can fit 22 keys
    // into 1,024 byte blocks. Therefore we will use a
    // tree of order 11 with 22 keys per bucket.
    KEY_SIZE = 41;
    // ORDER = 11;
     ORDER = 2; // For easier testing, make ORDER 2.
    KEYS_PER_NODE = 2 * ORDER;

    NIL_RECORD = -1; // A nil "pointer."

type
    // Record to hold header information or key buckets
    // in Custs.idx.
    TBucket = record
        case IsHeader : Boolean of
            True : ( // Header information.
                NumBuckets       : Longint; // # buckets in Custs.idx.
                NumRecords       : Longint; // # records in Custs.dat.
                Root             : Longint; // Index of root bucket in Custs.idx.
                NextTreeRecord   : Longint; // Next unused record in Custs.idx.
                NextCustRecord   : Longint; // Next unused record in Custs.dat.
                FirstTreeGarbage : Longint; // Top of garbage list in Custs.idx.
                FirstCustGarbage : Longint; // Top of garbage list in Custs.dat.
                Height           : Integer; // Height of tree.
            );
            False : ( // A bucket containing keys.
                NumKeys : Integer;                                      // # keys in use in this bucket.
                Key     : array [1..KEYS_PER_NODE] of String[KEY_SIZE]; // Key = Last Name, First Name.
                Child   : array [0..KEYS_PER_NODE] of Longint;          // Indexes of child buckets.
            );
    end; // End TBucket record definition.

    TBucketArray = array[1..1000000] of TBucket; // Array of buckets.
    PBucketArray = ^TBucketArray;                // Pointer to array of buckets.

    TIntArray = array [1..1000000] of Integer;   // Array of integers.
    PIntArray = ^TIntArray;                      // Pointer to array of integers.

    // Record for customer data in Custs.dat.
    TCustomer = record
        case IsGarbage : Boolean of
            True : (  // It's on the garbage list.
                NextGarbage : Longint; // Index of next record in garbage list.
            );
            False : ( // It's a real customer record.
                LastName   : String[20];
                FirstName  : String[20];
                Address    : String[40];
                City       : String[20];
                Voivodship : String[20];
                Zip        : String[10];
                Phone      : String[12];
            );
    end; // End TCustomer record definition.

    TBPlusTree = class(TObject)
        public
            Header         : TBucket;           // Cached header info.
            CachedBuckets  : PBucketArray;      // Array of cached buckets.
            CachedIndexes  : PIntArray;         // Indexes of cached buckets.
            IdxFile        : file of TBucket;   // Custs.idx file.
            DatFile        : file of TCustomer; // Custs.dat file.
            CurrentRecord  : TCustomer;         // Record currently selected.
            CurrentIndex   : Longint;           // Index of CurrentRecord.
            DiskAccesses   : Integer;           // # disk accesses this operation.

            FilesClosed    : Boolean;           // Are the data files closed?

            destructor Destroy; override;
            procedure OpenFiles(dir_name : String);
            procedure CloseFiles;
            procedure GetBucket(var buck : TBucket; bucknum : Longint);
            procedure CacheBucket(depth : Integer; bucknum : Longint);
            procedure PutBucket(buck : TBucket; bucknum : Longint);
            procedure FreeBucket(bucknum : Longint);
            procedure PutHeader;
            procedure GetCustomer(var cust : TCustomer; custnum : Longint);
            procedure PutCustomer(cust : TCustomer; custnum : Longint);
            procedure FreeCustomer(custnum : Longint);
            procedure Add;
            procedure AddToNode(depth : Integer; node : Longint; new_key : String; var up_node : Longint; var up_key : String; var had_split : Boolean);
            procedure AddToNodeWithRoom(depth : Integer; node : Longint; spot : Integer; var new_key : String; var new_child : Longint);
            procedure SplitBucket(depth : Integer; node : Longint; spot : Integer; var up_key : String; var up_node : Longint);
            function NewBucket(var buck : TBucket) : Longint;
            function NewCustomer(cust : TCustomer) : Longint;
            procedure FindRecord(key : String; var found_it : Boolean);
            procedure Remove;
            procedure RemoveFromNode(depth : Integer; key : String; var too_small : Boolean; var up_value : String);
            procedure TooSmall(var parent, small : TBucket; parent_index, small_index : Longint; spot : Integer; var too_small : Boolean);
            function TreeStructure(spaces : String; depth, lines : Integer; show_leaves : Boolean) : String;
    end;

implementation

// Close the data files and free dynamically allocated
// memory.
destructor TBPlusTree.Destroy;
begin
    // Free cache memory.
    if (Header.Height > 0) then
    begin
        // Free old cache memory.
        FreeMem(CachedBuckets);
        FreeMem(CachedIndexes);
    end;

    // Close the data files if necessary.
    if (not FilesClosed) then CloseFiles;

    inherited Destroy;
end;

// Open the indicated B+tree data files and cache the root
// bucket. If the files do not exist, create them and
// initialize the header information.
procedure TBPlusTree.OpenFiles(dir_name : String);
begin
    // Make sure the directory name ends in \.
    if (dir_name[Length(dir_name)] <> '\') then
        dir_name := dir_name + '\';

    // Open the files.
    AssignFile(IdxFile, dir_name + 'Custs.dat');
    AssignFile(DatFile, dir_name + 'Custs.idx');

    // Open the files, creating them if they don't exist.
    {$I-}
    Reset(IdxFile);
    if (IOResult = 0) then
    begin
        // The files exist.
        Reset(DatFile);

        // Read the header information.
        Read(IdxFile, Header);
    end else begin
        // The files do not exist. Create them.
        Rewrite(IdxFile);
        Rewrite(DatFile);

        // Create the header for an empty tree.
        with Header do
        begin
            IsHeader := True;
            NumBuckets := 0;
            NumRecords := 0;
            Root := NIL_RECORD;
            NextTreeRecord := 1; // First record after the header.
            NextCustRecord := 0; // First record in the file.
            FirstTreeGarbage := NIL_RECORD;
            FirstCustGarbage := NIL_RECORD;
            Height := 0;
        end;

        // Save the header info to the file.
        Write(IdxFile, Header);
    end; // End opening the files.
    {$I+}

    // Allocate the bucket cache.
    if (Header.Height > 0) then
    begin
        GetMem(CachedBuckets, Header.Height * SizeOf(TBucket));
        GetMem(CachedIndexes, Header.Height * SizeOf(Integer));

        // Read and cache the root bucket.
        GetBucket(CachedBuckets^[1], Header.Root);
        CachedIndexes^[1] := Header.Root;
    end;

    FilesClosed := False;
end;

// Close the B+tree data files.
procedure TBPlusTree.CloseFiles;
begin
    CloseFile(IdxFile);
    CloseFile(DatFile);
    FilesClosed := True;
end;

// Get a bucket from the file.
procedure TBPlusTree.GetBucket(var buck : TBucket; bucknum : Longint);
begin
    DiskAccesses := DiskAccesses + 1;
    Seek(IdxFile, bucknum);
    Read(IdxFile, buck);
end;

// Cache this bucket if it is not already cached.
procedure TBPlusTree.CacheBucket(depth : Integer; bucknum : Longint);
begin
    // If the bucket is already cached, do nothing.
    if (CachedIndexes^[depth] = bucknum) then exit;

    // Read the bucket.
    GetBucket(CachedBuckets^[depth], bucknum);
    CachedIndexes^[depth] := bucknum;
end;

// Save a bucket to file.
procedure TBPlusTree.PutBucket(buck : TBucket; bucknum : Longint);
begin
    DiskAccesses := DiskAccesses + 1;
    Seek(IdxFile, bucknum);
    Write(IdxFile, buck);
end;

// Move a customer record into the garbage list. The
// calling routine must save the header information
// to disk.
procedure TBPlusTree.FreeBucket(bucknum : Longint);
var
    buck : TBucket;
begin
    buck.Child[0] := Header.FirstTreeGarbage;
    PutBucket(buck, bucknum);
    Header.FirstTreeGarbage := bucknum;
    Header.NumBuckets := Header.NumBuckets - 1;
end;

// Save the header information to the file.
procedure TBPlusTree.PutHeader;
begin
    DiskAccesses := DiskAccesses + 1;
    Seek(IdxFile, 0);
    Write(IdxFile, Header);
end;

// Get a customer record from the file.
procedure TBPlusTree.GetCustomer(var cust : TCustomer; custnum : Longint);
begin
    DiskAccesses := DiskAccesses + 1;
    Seek(DatFile, custnum);
    Read(DatFile, cust);
end;

// Save a bucket to file.
procedure TBPlusTree.PutCustomer(cust : TCustomer; custnum : Longint);
begin
    DiskAccesses := DiskAccesses + 1;
    Seek(DatFile, custnum);
    Write(DatFile, cust);
end;

// Move a customer record into the garbage list. The
// calling routine must save the header information
// to disk.
procedure TBPlusTree.FreeCustomer(custnum : Longint);
var
    cust : TCustomer;
begin
    cust.NextGarbage := Header.FirstCustGarbage;
    PutCustomer(cust, custnum);
    Header.FirstCustGarbage := custnum;
    Header.NumRecords := Header.NumRecords - 1;
end;

// Add a new record to the tree.
procedure TBPlusTree.Add;
var
    up_node, old_root : Longint;
    key, up_key       : String;
    had_split         : Boolean;
begin
    DiskAccesses := 0; // Reset the access count.

    // Add the new item.
    had_split := False;
    key := CurrentRecord.LastName + ',' +
           CurrentRecord.FirstName;
    AddToNode(1, Header.Root, key, up_node, up_key, had_split);

    // If the root node split, create a new root containing
    // the item that was sent up to the parent.
    if (had_split) then
    begin
        // Make room for more cached buckets.
        if (Header.Height > 0) then
        begin
            // Free old cache memory.
            FreeMem(CachedBuckets);
            FreeMem(CachedIndexes);
        end;
        // Allocate new cache memory.
        GetMem(CachedBuckets, (Header.Height + 1) * SizeOf(TBucket));
        GetMem(CachedIndexes, (Header.Height + 1) * SizeOf(Integer));

        // Create the new root.
        old_root := Header.Root;
        CachedIndexes^[1] := NewBucket(CachedBuckets^[1]);
        with (CachedBuckets^[1]) do
        begin
            Key[1] := up_key;
            Child[0] := old_root;
            Child[1] := up_node;
            NumKeys := 1;
        end;

        // Save the new root to the index file.
        PutBucket(CachedBuckets^[1], CachedIndexes^[1]);

        // Prepare to update the header information.
        Header.Root := CachedIndexes^[1];
        Header.Height := Header.Height + 1;
    end; // End if we had a root split.

    // In either case the header was modified. At least
    // FirstCustGarbage has changed.
    PutHeader;
end;

// Recursively insert the item below the indicated node.
// If there is a split, add the item passed up to this
// node. If this node is then full, split it, too.
procedure TBPlusTree.AddToNode(depth : Integer; node : Longint; new_key : String; var up_node : Longint; var up_key : String; var had_split : Boolean);
var
    branch : Integer;
begin
    if (depth > Header.Height) then
    begin
        // We're at the bottom of the tree. Create the
        // new customer record and pass its record number
        // back up for insertion in the node above.
        CurrentIndex := NewCustomer(CurrentRecord);
        up_node := CurrentIndex;
        up_key := new_key;
        had_split := True;
        exit;
    end;

    // Cache the bucket.
    CacheBucket(depth, node);

    // See which branch we should investigate.
    for branch := 0 to CachedBuckets^[depth].NumKeys - 1 do
    begin
        if (CachedBuckets^[depth].Key[branch + 1] >
            new_key) then break;
    end;

    // Go down that branch.
    AddToNode(depth + 1,
        CachedBuckets^[depth].Child[branch], new_key,
        up_node, up_key, had_split);

    // If we had a split, deal with it.
    if (had_split) then
    begin
        if (CachedBuckets^[depth].NumKeys < KEYS_PER_NODE) then
        begin
            // If there's room, add the item here.
            AddToNodeWithRoom(depth, node, branch + 1, up_key, up_node);
            had_split := False;
        end else begin
            // The node is full. Split it.
            SplitBucket(depth, node, branch + 1, up_key, up_node);
        end;
    end;
end;

// Insert an item in a node known to have a room for it.
procedure TBPlusTree.AddToNodeWithRoom(depth : Integer; node : Longint; spot : Integer; var new_key : String; var new_child : Longint);
var
    i : Integer;
begin
    // Make room for the new entry.
    with CachedBuckets^[depth] do
    begin
        NumKeys := NumKeys + 1;
        for i := NumKeys downto spot + 1 do
        begin
            Key[i] := Key[i - 1];
            Child[i] := Child[i - 1];
        end;

        // Insert the new entry.
        Key[spot] := new_key;
        Child[spot] := new_child;
    end; // End with CachedBuckets^[depth] do ...

    // Update the record to disk.
    PutBucket(CachedBuckets^[depth], CachedIndexes^[depth]);
end;

// Split the node plus the up_key item into two nodes.
// Variable spot indicates where in the node the up_key
// item should be placed.
//
// Return the new node through parameter up_node. Return
// the key to be passed back up through parameter up_key.
procedure TBPlusTree.SplitBucket(depth : Integer; node : Longint; spot : Integer; var up_key : String; var up_node : Longint);
var
    i                          : Integer;
    return_node, right_child_0 : Longint;
    return_bucket              : TBucket;
    return_key                 : String;
begin
    with CachedBuckets^[depth] do
    begin
        // Create a new bucket to the right.
        return_node := NewBucket(return_bucket);

        if (spot <= ORDER + 1) then
        begin
            // The new item belongs in the left node or
            // should be passed up.
            if (spot = ORDER + 1) then
            begin
                // Pass it up.
                return_key := up_key;
                right_child_0 := up_node;
            end else begin
                // Save the ORDER-th entry.
                return_key := Key[ORDER];
                right_child_0 := Child[ORDER];

                // Insert the new item into the left node.
                for i := ORDER downto spot + 1 do
                begin
                    Key[i] := Key[i - 1];
                    Child[i] := Child[i - 1];
                end;
                Key[spot] := up_key;
                Child[spot] := up_node;
            end; // End with CachedBuckets^[depth] do

            // Move items into the right node.
            for i := 1 to ORDER do
            begin
                return_bucket.Key[i] := Key[i + ORDER];
                return_bucket.Child[i] := Child[i + ORDER];
            end;
        end else begin
            // The new item should be in the right node.
            spot := spot - ORDER - 1;

            return_key := Key[ORDER + 1];
            right_child_0 := Child[ORDER + 1];

            // Move the first set of items into the right node.
            for i := 1 to spot - 1 do
            begin
                return_bucket.Key[i] := Key[i + ORDER + 1];
                return_bucket.Child[i] := Child[i + ORDER + 1];
            end;

            // Put the up_key/up_node in the right node.
            return_bucket.Key[spot] := up_key;
            return_bucket.Child[spot] := up_node;

            // Put the other items in the right node.
            for i := spot + 1 to ORDER do
            begin
                return_bucket.Key[i] := Key[i + ORDER];
                return_bucket.Child[i] := Child[i + ORDER];
            end;
        end; // End (in left) ... else (in right) ...

        NumKeys := ORDER;
        return_bucket.NumKeys := ORDER;
        return_bucket.Child[0] := right_child_0;
        up_node := return_node;
        up_key := return_key;
    end; // End with CachedBuckets^[depth] do

    // Update the split buckets.
    PutBucket(return_bucket, return_node);
    PutBucket(CachedBuckets^[depth], CachedIndexes^[depth]);
end;

// Return the record number for an unused B+Tree bucket.
// The calling routine must save the updated header
// information and the new bucket to the disk.
function TBPlusTree.NewBucket(var buck : TBucket) : Longint;
begin
    if (Header.FirstTreeGarbage = NIL_RECORD) then
    begin
        // The garbage list is empty. Take a new bucket
        // from the end of the file.
        Result := Header.NextTreeRecord;
        Header.NextTreeRecord := Header.NextTreeRecord + 1;
    end else begin
        // Reuse the top garbage item.
        Result := Header.FirstTreeGarbage;
        GetBucket(buck, Result);
        Header.FirstTreeGarbage := buck.Child[0];
    end;
    Header.NumBuckets := Header.NumBuckets + 1;

    // Blank the new bucket.
    buck.NumKeys := 0;
end;

// Return the record number for an unused customer record.
// Fill in the fields from cust. The calling routine must
// save the updated header information.
function TBPlusTree.NewCustomer(cust : TCustomer) : Longint;
var
    garbage : TCustomer;
begin
    // See if there is one on the garbage list we can use.
    if (Header.FirstCustGarbage = NIL_RECORD) then
    begin
        // The garbage list is empty, take a new record
        // from the end of the file.
        Result := Header.NextCustRecord;
        Header.NextCustRecord := Header.NextCustRecord + 1;
    end else begin
        // Reuse the top garbage record.
        Result := Header.FirstCustGarbage;
        GetCustomer(garbage, Result);
        Header.FirstCustGarbage := garbage.NextGarbage;
    end;
    Header.NumRecords := Header.NumRecords + 1;

    // Write the data in rec to the file.
    PutCustomer(cust, Result);
end;

// Find the indicated record.
procedure TBPlusTree.FindRecord(key : String; var found_it : Boolean);
var
    depth, branch : Integer;
    node          : Longint;
begin
    DiskAccesses := 0; // Reset the access count.

    // If the tree is empty, do nothing.
    if (Header.Height < 1) then
    begin
        found_it := False;
        CurrentIndex := NIL_RECORD;
        exit;
    end;

    // During this search, node is the index in Custs.idx
    // of the bucket we are searching. Depth is our depth
    // in the tree. It is also the entry in Buckets that
    // holds the bucket we are examining. We will move
    // down the tree until we get to a leaf.

    // Start at the cached root.
    for depth := 1 to Header.Height - 1 do
    begin
        // See down which branch we should move.
        for branch := 0 to CachedBuckets^[depth].NumKeys - 1 do
            if (CachedBuckets^[depth].Key[branch + 1] > key)
                then break;

        // Move down the branch.
        node := CachedBuckets^[depth].Child[branch];
        CachedIndexes^[depth + 1] := node;
        GetBucket(CachedBuckets^[depth + 1], node);
    end;

    // See where the actual data lies.
    depth := Header.Height;
    for branch := 0 to CachedBuckets^[depth].NumKeys - 1 do
        if (CachedBuckets^[depth].Key[branch + 1] > key)
            then break;

    // See if we found the data.
    found_it := True;
    if (CachedBuckets^[depth].Child[branch] = NIL_RECORD) then
    begin
        found_it := False
    end else if (branch > 0) then
    begin
        if (key <> CachedBuckets^[depth].Key[branch]) then
            found_it := False;
    end;

    // If we have a match, get the data.
    if (found_it) then
    begin
        // Get the record.
        CurrentIndex := CachedBuckets^[depth].Child[branch];
        GetCustomer(CurrentRecord, CurrentIndex);
    end else begin
        // Unselect the current record.
        CurrentIndex := NIL_RECORD;
    end;
end;

// Remove the current record.
procedure TBPlusTree.Remove;
var
    too_small : Boolean;
    old_root  : Longint;
    up_value  : String;
begin
    DiskAccesses := 0; // Reset the access count.

    // If the tree is empty, do nothing.
    if (Header.Height < 1) then
    begin
        CurrentIndex := NIL_RECORD;
        exit;
    end;

    // Delete the item.
    RemoveFromNode(1,
        CurrentRecord.LastName + ',' +
        CurrentRecord.FirstName,
        too_small, up_value);
    CurrentIndex := NIL_RECORD;

    // If the root shrunk, see if we need to remove a
    // level from the tree.
    if (CachedBuckets^[1].NumKeys < 1) then
    begin
        // Remember the old root so we can free it.
        old_root := Header.Root;

        // Update the header.
        Header.Height := Header.Height - 1;
        Header.Root := CachedBuckets^[1].Child[0];

        // Free the old root.
        FreeBucket(old_root);

        // See if the tree is empty.
        if (Header.Height > 0) then
        begin
            // The tree isn't empty. Cache the new root.
            CacheBucket(1, Header.Root);
        end else begin
            // Otherwise reset everything.
            with Header do
            begin
                IsHeader := True;
                NumBuckets := 0;
                NumRecords := 0;
                Root := NIL_RECORD;
                NextTreeRecord := 1; // First record after the header.
                NextCustRecord := 0; // First record in the file.
                FirstTreeGarbage := NIL_RECORD;
                FirstCustGarbage := NIL_RECORD;
                Height := 0;
            end;
        end;
    end; // End if the root shrunk...

    // In either case the header was modified. At least
    // FirstCustGarbage has changed.
    PutHeader;
end;

// Recursively remove the item from somewhere below
// this node.
procedure TBPlusTree.RemoveFromNode(depth : Integer; key : String; var too_small : Boolean; var up_value : String);
var
    branch, i : Integer;
begin
    // See down which branch we should search.
    for branch := 1 to CachedBuckets^[depth].NumKeys do
        if (CachedBuckets^[depth].Key[branch] > key) then
            break;

    // If we are at a leaf, this better be it.
    if (depth >= Header.Height) then
    begin
        // Make depth point to the customer record which
        // should match the target.
        branch := branch - 1;

        // Get the customer record and see if it matches.
        CurrentIndex := CachedBuckets^[depth].Child[branch];
        if (CurrentIndex <> NIL_RECORD) then
        begin
            GetCustomer(CurrentRecord, CurrentIndex);
            if (CurrentRecord.LastName + ',' +
                CurrentRecord.FirstName <> key) then
                    CurrentIndex := NIL_RECORD;
        end;
        if (CurrentIndex = NIL_RECORD) then
        begin
            // The record is not here.
            too_small := False;
            exit;
        end;

        // We have verified that this is the target.
        // Since this is a B+Tree, we know we are at a
        // leaf. Simply remove the leaf entry and return
        // up the tree.

        // If this is the leftmost entry in the leaf, we
        // must revise the ancestor that holds the target
        // value as a key.
        if (branch = 0) then
            up_value := CachedBuckets^[depth].Key[1]
        else
            up_value := '';

        // Remove the entry from the leaf node.
        with CachedBuckets^[depth] do
        begin
            NumKeys := NumKeys - 1;
            too_small := (NumKeys < ORDER);

            if (branch <= NumKeys) Then
                Child[branch] := Child[branch + 1];
            if (branch > 0) then
                Key[branch] := Key[branch + 1];

            for i := branch + 1 to NumKeys do
            begin
                Key[i] := Key[i + 1];
                Child[i] := Child[i + 1];
            end;
        end; // End with CachedBuckets^[depth] do ...

        // Save the updated leaf node to disk.
        PutBucket(CachedBuckets^[depth], CachedIndexes^[depth]);

        // Free the customer record.
        FreeCustomer(CurrentIndex);

        // We are done performing the deletion.
        exit;
    end;

    // We have not yet found the item. Continue down the
    // tree looking for it.

    // Cache the next bucket in the search.
    CurrentIndex := CachedBuckets^[depth].Child[branch - 1];
    CacheBucket(depth + 1, CurrentIndex);

    // Delete the item from that bucket.
    RemoveFromNode(depth + 1, key, too_small, up_value);

    // If this node holds the exact value, update it with
    // up_value.
    if ((branch > 1) and (up_value <> '')) then
    begin
        if (CachedBuckets^[depth].Key[branch - 1] = key) then
        begin
            CachedBuckets^[depth].Key[branch - 1] := up_value;
            up_value := '';
        end;
    end;

    // If the child node is now too small, rearrange it.
    if (too_small) then
        TooSmall(
            CachedBuckets^[depth],
            CachedBuckets^[depth + 1],
            CachedIndexes^[depth],
            CachedIndexes^[depth + 1],
            branch - 1, too_small);
end;

// Node small is too small. It is in position spot in
// parent node parent.
//
// If node has a right sibling, try to redistribute items
// between the two. If the right sibling has too few items
// (it has exactly ORDER items), merge them.
//
// If node has no right sibling, try the left sibling.
//
// If two nodes merge, see if the parent is now too small.
//
// The calling routine must save the header information
// to disk. It will be updated if a bucket is freed.
procedure TBPlusTree.TooSmall(var parent, small : TBucket; parent_index, small_index : Longint; spot : Integer; var too_small : Boolean);
var
    num_in_parent, num_in_sibling : Integer;
    num_to_move, i                : Integer;
    node, sib_index               : Longint;
    sibling                       : TBucket;
begin
    num_in_parent := parent.NumKeys;

    // See if there is a right sibling.
    if (spot < num_in_parent) then
    begin
        // Examine the right sibling.
        spot := spot + 1; // Now points to the sibling.
        sib_index := parent.Child[spot]; // Sibling's index.
        GetBucket(sibling, sib_index);
        num_in_sibling := sibling.NumKeys;
        num_to_move := (num_in_sibling - ORDER + 1) div 2;

        // Put item from parent at end of items in the
        // small node. Right now it has ORDER - 1 keys.
        small.Key[ORDER] := parent.Key[spot];
        small.Child[ORDER] := sibling.Child[0];

        if (num_to_move > 0) then
        begin
            // If we have enough items to redistribute,
            // do so. Copy items from sibling to small.
            for i := 1 to num_to_move - 1 do
            begin
                small.Key[i + ORDER] := sibling.Key[i];
                small.Child[i + ORDER] := sibling.Child[i];
            end;

            // Prepare the parent.
            parent.Key[spot] := sibling.Key[num_to_move];
            parent.Child[spot] := sib_index;

            // Fill the hole in the sibling.
            sibling.Child[0] := sibling.Child[num_to_move];
            num_in_sibling := num_in_sibling - num_to_move;
            for i := 1 to num_in_sibling do
            begin
                sibling.Key[i] := sibling.Key[i + num_to_move];
                sibling.Child[i] := sibling.Child[i + num_to_move];
            end;

            // Update the key counts.
            sibling.NumKeys := num_in_sibling;
            small.NumKeys := ORDER - 1 + num_to_move;
            too_small := False;

            // Save the updated buckets to disk.
            PutBucket(parent, parent_index);
            PutBucket(small, small_index);
            PutBucket(sibling, sib_index);
        end else begin
            // There are not enough items to redistribute.
            // Merge the nodes.

            // Move entries from sibling to small. It must
            // contain ORDER items.
            for i := 1 to ORDER do
            begin
                small.Key[i + ORDER] := sibling.Key[i];
                small.Child[i + ORDER] := sibling.Child[i];
            end;

            // Fill in the hole in the parent.
            for i := spot to num_in_parent - 1 do
            begin
                parent.Key[i] := parent.Key[i + 1];
                parent.Child[i] := parent.Child[i + 1];
            end;

            // Update the key counts.
            small.NumKeys := KEYS_PER_NODE;
            parent.NumKeys := num_in_parent - 1;

            // Free the sibling node.
            FreeBucket(sib_index);
            too_small := (parent.NumKeys < ORDER);

            // Save the updated buckets to disk.
            PutBucket(parent, parent_index);
            PutBucket(small, small_index);
        end; // End if (can redistribute) ... else (merge) ...
    end else begin
        // There is no right sibling. Look at the left
        // sibling.
        sib_index := parent.Child[spot - 1];
        GetBucket(sibling, sib_index);
        num_in_sibling := sibling.NumKeys + 1;
        num_to_move := (num_in_sibling - ORDER) div 2;
        if (num_to_move > 0) then
        begin
            // We have enough items to redistribute. Do so.

            // Make room for the new items in the node.
            for i := ORDER - 1 downto 1 do
            begin
                small.Key[i + num_to_move] := small.Key[i];
                small.Child[i + num_to_move] := small.Child[i];
            end;

            // Move an item from the parent into the node.
            small.Key[num_to_move] := parent.Key[spot];
            small.Child[num_to_move] := small.Child[0];

            // Move items from the sibling into the node.
            num_in_sibling := num_in_sibling - num_to_move;
            for i := num_to_move - 1 downto 1 do
            begin
                small.Key[i] := sibling.Key[i + num_in_sibling];
                small.Child[i] := sibling.Child[i + num_in_sibling];
            end;
            small.Child[0] := sibling.Child[num_in_sibling];

            // Revise the sibling's entry in the parent.
            parent.Key[spot] := sibling.Key[num_in_sibling];

            // Update the key counts.
            sibling.NumKeys := num_in_sibling - 1;
            small.NumKeys := ORDER - 1 + num_to_move;
            too_small := False;

            // Save the updated buckets to disk.
            PutBucket(parent, parent_index);
            PutBucket(small, small_index);
            PutBucket(sibling, sib_index);
        end else begin
            // Not enough items to redistribute. Merge.

            // Put the node's parent entry in the sibling.
            sibling.Key[num_in_sibling] := parent.Key[spot];
            sibling.Child[num_in_sibling] := small.Child[0];

            // Move items from the node into the sibling.
            for i := 1 to ORDER - 1 do
            begin
                sibling.Key[i + num_in_sibling] := small.Key[i];
                sibling.Child[i + num_in_sibling] := small.Child[i];
            end;

            // Update the key counts.
            sibling.NumKeys := KEYS_PER_NODE;
            parent.NumKeys := num_in_parent - 1;

            // Free the node.
            FreeBucket(small_index);
            too_small := (parent.NumKeys < ORDER);

            // Save the updated buckets to disk.
            PutBucket(parent, parent_index);
            PutBucket(sibling, sib_index);
        end; // End if (can redistribute) ... else (merge) ...
    end; // End if (right sibling) ... else (left sibling) ...
end;

// Return a textual representation of the tree's structure.
function TBPlusTree.TreeStructure(spaces : String; depth, lines : Integer; show_leaves : Boolean) : String;
const
    MAX_LINES = 500;
    INDENT = '    ';
    CR = #13#10;
var
    node : Longint;
    i    : Integer;
    cust : TCustomer;
begin
    if (lines > MAX_LINES) then exit;
    Result := '';
    if (depth < Header.Height) then
    begin
        // We are in an internal node.
        node := CachedBuckets^[depth].Child[0];
        CachedIndexes^[depth + 1] := node;
        GetBucket(CachedBuckets^[depth + 1], node);
        Result := TreeStructure(spaces + INDENT, depth + 1, lines, show_leaves);
        if (lines > MAX_LINES) then exit;

        // Handle the keys and other branches.
        for i := 1 to CachedBuckets^[depth].NumKeys do
        begin
            // Display the key.
            Result := Result + spaces +
                CachedBuckets^[depth].Key[i] + CR;
            lines := lines + 1;
            if (lines > MAX_LINES) then
            begin
                Result := Result + CR + '[itd.]';
                exit;
            end;

            // Head down the corresponding child pointer.
            node := CachedBuckets^[depth].Child[i];
            CachedIndexes^[depth + 1] := node;
            GetBucket(CachedBuckets^[depth + 1], node);
            Result := Result +
                TreeStructure(spaces + INDENT, depth + 1, lines, show_leaves);
            if (lines > MAX_LINES) then exit;
        end; // End for i := 1 to CachedBuckets^[depth].NumKeys do
    end else begin
        // We have reached a leaf.
        node := CachedBuckets^[depth].Child[0];
        if (show_leaves) then
        begin
            if (node = NIL_RECORD) then
            begin
                Result := Result + spaces + INDENT +
                    '<NO_CHILD>' + CR
            end else begin
                GetCustomer(cust, node);
                Result := Result + spaces + INDENT +
                    '<' + cust.LastName + ',' +
                    cust.FirstName + '>' + CR;
            end;
            lines := lines + 1;
            if (lines > MAX_LINES) then
            begin
                Result := Result + CR + '[itd.]';
                exit;
            end;
        end;

        // Handle the keys and other customer records.
        for i := 1 to CachedBuckets^[depth].NumKeys do
        begin
            // Display the key.
            Result := Result + spaces +
                CachedBuckets^[depth].Key[i] + CR;
            lines := lines + 1;
            if (lines > MAX_LINES) then
            begin
                Result := Result + CR + '[itd.]';
                exit;
            end;

            // Head down the corresponding child pointer.
            node := CachedBuckets^[depth].Child[i];
            if (show_leaves) then
            begin
                if (node = NIL_RECORD) then
                begin
                    Result := Result + spaces + INDENT +
                        '<NO_CHILD>' + CR
                end else begin
                    GetCustomer(cust, node);
                    Result := Result + spaces + INDENT +
                        '<' + cust.LastName + ',' +
                        cust.FirstName + '>' + CR;
                end;
                lines := lines + 1;
                if (lines > MAX_LINES) then
                begin
                    Result := Result + CR + '[itd.]';
                    exit;
                end;
            end;
        end; // End for i := 1 to CachedBuckets^[depth].NumKeys do
    end; // End if (internal node) ... ellse (leaf node) ...
end;

end.

