dbf_idxfile.pas

tDBF is new ver, this is BDS 2007 insta

  // inv1: (ARecNo<>-3) -> Entry(low-1).Key <  Key <= Entry(high).Key
  // inv2: (ARecNo =-3) -> Entry(low-1).Key <= Key <  Entry(high).Key
  // vf: high + 1 - low
  while low < high do
  begin
    current := (low + high) div 2;
    FEntry := GetEntry(current);
    // calc diff
    Result := MatchKey;
    // test if we need to go lower or higher
    // result < 0 implies key smaller than tested entry
    // result = 0 implies key equal to tested entry
    // result > 0 implies key greater than tested entry
    if (Result < 0) or ((ARecNo<>-3) and (Result=0)) then
      high := current
    else
      low := current+1;
  end;
  // high will contain first greater-or-equal key
  // ARecNo <> -3 -> Entry(high).Key will contain first key that matches    -> go to high
  // ARecNo =  -3 -> Entry(high).Key will contain first key that is greater -> go to high
  FEntryNo := -1;
  EntryNo := high;
  // calc end result: can't inspect high if lowerpage <> nil
  // if this is a leaf, we need to find specific recno
  if (LowerPage = nil) then
  begin
    if high > FHighIndex then
    begin
      Result := 1;
    end else begin
      Result := MatchKey;
      // test if we need to find a specific recno
      // result < 0 -> current key greater -> nothing found -> don't search
      if (ARecNo > 0) then
      begin
        // BLS to RecNo
        high := FHighIndex + 1;
        low := FEntryNo;
        // inv: FLowIndex <= FEntryNo <= high <= FHighIndex + 1 /\
        // (Ai: FLowIndex <= i < FEntryNo: Entry(i).RecNo <> ARecNo)
        while FEntryNo <> high do
        begin
          // FEntryNo < high, get new entry
          if low <> FEntryNo then
          begin
            FEntry := GetEntry(FEntryNo);
            // check if entry key still ok
            Result := MatchKey;
          end;
          // test if out of range or found recno
          if (Result <> 0) or (GetRecNo = ARecNo) then
            high := FEntryNo
          else begin
            // default to EOF
            inc(FEntryNo);
            Result := 1;
          end;
        end;
      end;
    end;
  end else begin
    // FLowerPage <> nil -> high contains entry, can not have empty range
    Result := 0;
  end;
end;

procedure TIndexPage.GotoInsertEntry;
  // assures we really can insert here
begin
  if FEntry = FIndexFile.EntryEof then
    FEntry := GetEntry(FEntryNo);
end;

procedure TIndexPage.SetEntry(RecNo: Integer; AKey: PChar; LowerPageNo: Integer);
var
  keyData: PChar;
{$ifdef TDBF_INDEX_CHECK}
  prevKeyData, curKeyData, nextKeyData: PChar;
{$endif}
begin
  // get num entries
  keyData := GetKeyData;
  // check valid entryno: we should be able to insert entries!
  assert((EntryNo >= 0) and (EntryNo <= FHighIndex));
  if (UpperPage <> nil) and (FEntryNo = FHighIndex) then
    UpperPage.SetEntry(0, AKey, FPageNo);
{  if PIndexHdr(FIndexFile.IndexHeader).KeyType = 'C' then  }
    if AKey <> nil then
      Move(AKey^, keyData^, SwapWordLE(PIndexHdr(FIndexFile.IndexHeader)^.KeyLen))
    else
      PChar(keyData)^ := #0;
{
  else
    if AKey <> nil then
      PDouble(keyData)^ := PDouble(AKey)^
    else
      PDouble(keyData)^ := 0.0;
}
  // set entry info
  SetRecLowerPageNo(RecNo, LowerPageNo);
  // flag we modified the page
  FModified := true;

{$ifdef TDBF_INDEX_CHECK}

    // check sorted entry sequence
    prevKeyData := GetKeyDataFromEntry(FEntryNo-1);
    curKeyData  := GetKeyDataFromEntry(FEntryNo+0);
    nextKeyData := GetKeyDataFromEntry(FEntryNo+1);
    // check if prior entry not greater, 'rightmost' key does not have to match
    if (FEntryNo > 0) and ((FLowerPage = nil) or (FEntryNo < FHighIndex)) then
    begin
      if FIndexFile.CompareKeys(prevKeyData, curKeyData) > 0 then
        assert(false);
    end;
    // check if next entry not smaller
    if ((FLowerPage = nil) and (FEntryNo < FHighIndex)) or
        ((FLowerPage <> nil) and (FEntryNo < (FHighIndex - 1))) then
    begin
      if FIndexFile.CompareKeys(curKeyData, nextKeyData) > 0 then
        assert(false);
    end;

{$endif}

end;

{$ifdef TDBF_UPDATE_FIRST_LAST_NODE}

procedure TIndexPage.SetPrevBlock(NewBlock: Integer);
begin
end;

{$endif}

procedure TIndexPage.Split;
  // *) assumes this page is `nearly' full
var
  NewPage: TIndexPage;
  source, dest: Pointer;
  paKeyData: PChar;
  size, oldEntryNo: Integer;
  splitRight, lNumEntries, numEntriesNew: Integer;
  saveLow, saveHigh: Integer;
  newRoot: Boolean;
begin
  // assure parent exists, if not -> create & lock, else lock it
  newRoot := FUpperPage = nil;
  if newRoot then
    FIndexFile.AddNewLevel
  else
    FUpperPage.LockPage;

  // lock this page for updates
  LockPage;

  // get num entries
  lNumEntries := GetNumEntries;

  // calc split pos: split in half
  splitRight := lNumEntries div 2;
  if (FLowerPage <> nil) and (lNumEntries mod 2 = 1) then
    inc(splitRight);
  numEntriesNew := lNumEntries - splitRight;
  // check if place to insert has least entries
  if (numEntriesNew > splitRight) and (EntryNo > splitRight) then
  begin
    inc(splitRight);
    dec(numEntriesNew);
  end else if (numEntriesNew < splitRight) and (EntryNo < splitRight) then
  begin
    dec(splitRight);
    inc(numEntriesNew);
  end;
  // save current entryno
  oldEntryNo := EntryNo;
  // check if we need to save high / low bound
  if FLowPage = FPageNo then
    saveLow := FLowIndex
  else
    saveLow := -1;
  if FHighPage = FPageNo then
    saveHigh := FHighIndex
  else
    saveHigh := -1;

  // create new page
  NewPage := TIndexPageClass(ClassType).Create(FIndexFile);
  try
    // get page
    NewPage.GetNewPage;
{$ifdef TDBF_UPDATE_FIRSTLAST_NODE}
    NewPage.SetPrevBlock(NewPage.PageNo - FIndexFile.PagesPerRecord);
{$endif}

    // set modified
    FModified := true;
    NewPage.FModified := true;

    // compute source, dest
    dest := NewPage.GetEntry(0);
    source := GetEntry(splitRight);
    size := numEntriesNew * SwapWordLE(PIndexHdr(FIndexFile.IndexHeader)^.KeyRecLen);
    // if inner node, copy rightmost entry too
    if FLowerPage <> nil then
      size := size + FIndexFile.EntryHeaderSize;
    // copy bytes
    Move(source^, dest^, size);
    // if not inner node, clear possible 'rightmost' entry
    if (FLowerPage = nil) then
      SetRecLowerPageNoOfEntry(splitRight, 0, 0);

    // calc new number of entries of this page
    lNumEntries := lNumEntries - numEntriesNew;
    // if lower level, then we need adjust for new 'rightmost' node
    if FLowerPage <> nil then
    begin
      // right split, so we need 'new' rightmost node
      dec(lNumEntries);
    end;
    // store new number of nodes
    // new page is right page, so update parent to point to new right page
    NewPage.SetNumEntries(numEntriesNew);
    SetNumEntries(lNumEntries);
    // update highindex
    FHighIndex := lNumEntries;
    if FLowerPage = nil then
      dec(FHighIndex);

    // get data of last entry on this page
    paKeyData := GetKeyDataFromEntry(splitRight - 1);

    // reinsert ourself into parent
//    FUpperPage.RecurInsert(0, paKeyData, FPageNo);
    // we can do this via a localinsert now: we know there is at least one entry
    // free in this page and higher up
    FUpperPage.LocalInsert(0, paKeyData, FPageNo);

    // new page is right page, so update parent to point to new right page
    // we can't do this earlier: we will get lost in tree!
    FUpperPage.SetRecLowerPageNoOfEntry(FUpperPage.EntryNo+1, 0, NewPage.PageNo);

    // NOTE: UpperPage.LowerPage = Self <= inserted FPageNo, not NewPage.PageNo
  finally
    NewPage.Free;
  end;

  // done updating: unlock page
  UnlockPage;
  // save changes to parent
  FUpperPage.UnlockPage;

  // unlock new root, unlock header too
  FIndexFile.UnlockHeader;

  // go to entry we left on
  if oldEntryNo >= splitRight then
  begin
    // sync upperpage with right page
    FUpperPage.EntryNo := FUpperPage.EntryNo + 1;
    FEntryNo := oldEntryNo - splitRight;
    FEntry := GetEntry(FEntryNo);
  end else begin
    // in left page = this page
    EntryNo := oldEntryNo;
  end;

  // check if we have to save high / low bound
  // seen the fact that FHighPage = FPageNo -> EntryNo <= FHighIndex, it can in
  // theory not happen that page is advanced to right page and high bound remains
  // on left page, but we won't check for that here
  if saveLow >= splitRight then
  begin
    FLowPage := FPageNo;
    FLowIndex := saveLow - splitRight;
  end;
  if saveHigh >= splitRight then
  begin
    FHighPage := FPageNo;
    FHighIndex := saveHigh - splitRight;
  end;
end;

procedure TIndexPage.Delete;
begin
  LocalDelete;
end;

procedure TIndexPage.WritePage;
begin
  // check if we modified current page
  if FModified and (FPageNo > 0) then
  begin
    FIndexFile.WriteRecord(FPageNo, FPageBuffer);
    FModified := false;
  end;
end;

procedure TIndexPage.Flush;
begin
  WritePage;
  if FLowerPage <> nil then
    FLowerPage.Flush;
end;

procedure TIndexPage.RecalcWeight;
begin
  if FLowerPage <> nil then
  begin
    FWeight := FLowerPage.Weight * SwapWordLE(PIndexHdr(FIndexFile.IndexHeader)^.NumKeys);
  end else begin
    FWeight := 1;
  end;
  if FUpperPage <> nil then
    FUpperPage.RecalcWeight;
end;

procedure TIndexPage.UpdateWeight;
begin
  if FLowerPage <> nil then
    FLowerPage.UpdateWeight
  else
    RecalcWeight;
end;

procedure TIndexPage.SetUpperPage(NewPage: TIndexPage);
begin
  if FUpperPage <> NewPage then
  begin
    // root height changed: update weights
    FUpperPage := NewPage;
    UpdateWeight;
  end;
end;

procedure TIndexPage.SetLowPage(NewPage: Integer);
begin
  if FLowPage <> NewPage then
  begin
    FLowPage := NewPage;
    UpdateBounds(FLowerPage <> nil);
  end;
end;

procedure TIndexPage.SetHighPage(NewPage: Integer);
begin
  if FHighPage <> NewPage then
  begin
    FHighPage := NewPage;
    UpdateBounds(FLowerPage <> nil);
  end;
end;

procedure TIndexPage.UpdateBounds(IsInnerNode: Boolean);
begin
  // update low / high index range
  if FPageNo = FLowPage then
    FLowIndex := FLowBracket
  else
    FLowIndex := 0;
  if FPageNo = FHighPage then
    FHighIndex := FHighBracket
  else begin
    FHighIndex := GetNumEntries;
    if not IsInnerNode then
      dec(FHighIndex);
  end;
end;

function TMdxPage.GetIsInnerNode: Boolean;
begin
  Result := SwapIntLE(PMdxPage(FPageBuffer)^.NumEntries) < SwapWordLE(PIndexHdr(FIndexFile.IndexHeader)^.NumKeys);
  // if there is still an entry after the last one, this has to be an inner node
  if Result then
    Result := PMdxEntry(GetEntry(PMdxPage(FPageBuffer)^.NumEntries))^.RecBlockNo <> 0;
end;

function TNdxPage.GetIsInnerNode: Boolean;
begin
  Result := PNdxEntry(GetEntry(0))^.LowerPageNo <> 0;
end;

procedure TIndexPage.SetPageNo(NewPageNo: Integer);
var
  isInnerNode: Boolean;
begin
  if (NewPageNo <> FPageNo) or FIndexFile.NeedLocks then
  begin
    // save changes
    WritePage;
    // no locks
    assert(FLockCount = 0);

    // goto new page
    FPageNo := NewPageNo;
    // remind ourselves we need to load new entry when page loaded
    FEntryNo := -1;
    if (NewPageNo > 0) and (NewPageNo <= FIndexFile.RecordCount) then
    begin
      // read page from disk
      FIndexFile.ReadRecord(NewPageNo, FPageBuffer);

      // fixup descending tree
      isInnerNode := GetIsInnerNode;

      // update low / high index range
      UpdateBounds(isInnerNode);

      // read inner node if any
      if isInnerNode then
      begin
        if FLowerPage = nil then
        begin
          FLowerPage := TIndexPageClass(ClassType).Create(FIndexFile);
          FLowerPage.UpperPage := Self;
        end;
        // read first entry, don't do this sooner, not created lowerpage yet
        // don't recursively resync all lower pages
{$ifdef TDBF_INDEX_CHECK}
      end else if FLowerPage <> nil then
      begin
//        FLowerPage.Free;
//        FLowerPage := nil;
        assert(false);
{$endif}
      end else begin
        // we don't have to check autoresync here because we're already at lowest level
        EntryNo := FLowIndex;
      end;
    end;
  end;
end;

procedure TIndexPage.SyncLowerPage;