!=========================================================================
! Copyright (C) VMware, Inc. 1986-2011.  All Rights Reserved.
!
! $Id: btreeinterior.gs,v 1.22 2008-01-09 22:50:08 stever Exp $
!
! Superclass Hierarchy:
!   BtreeInteriorNode, BtreeNode, Array, SequenceableCollection,
!   Collection, Object.
!
!=========================================================================

! class created in idxclasses.topaz

removeallmethods BtreeInteriorNode
removeallclassmethods BtreeInteriorNode

category: 'For Documentation Installation only'
classmethod: BtreeInteriorNode
installDocumentation

| doc txt |
doc := GsClassDocumentation _newForPrivateGsClass: self.

txt := (GsDocText new) details:
'A BtreeInteriorNode represents an interior node within a B-tree.' .
doc documentClassWith: txt.

txt := (GsDocText new) details:
'The value of the last element contained in the node.'.
doc documentInstVar: #lastValue with: txt.

self description: doc.
%

category: 'Removing'
method: BtreeInteriorNode
_attemptTakingFromNeighborForNodeAt: nodeIndex

"The node at the given index has too few entries, so try to take entries from
 one of its neighboring siblings.  First try the next (right) sibling, then
 the previous (left) sibling."

| node sibling eSize siblingLoss numSiblingElements copyStart copyEnd
fromNextSibling fromPreviousSibling targetStart lastChildKey 
siblingLastIdx numToMove |

node := super at: nodeIndex.
fromNextSibling := false.  " is true if taken from next sibling "
fromPreviousSibling := false.  " is true if taken from previous sibling "

eSize := node entrySize.
" check if the node is the last entry "
nodeIndex == (self _lastKeyIndex - 1)
    ifFalse: [ " this is not the last entry, check the next sibling "
        sibling := super at: (nodeIndex + self entrySize).
        numSiblingElements := sibling numElements.
        numSiblingElements < (node mergeThreshold + 1)
            ifFalse: [ " can take entries from next sibling "
                fromNextSibling := true.
                siblingLoss := (numSiblingElements - node mergeThreshold) quo: 2.
                siblingLoss == 0 ifTrue: [ siblingLoss := 1 ].
                copyStart := 1.
                copyEnd := siblingLoss * eSize.
                numToMove := 0 .
                targetStart := node _lastIndex + 1
            ]
    ].

" if could not get entries from next sibling and this is not the first entry "
(fromNextSibling not _and: [ nodeIndex > 1 ] )
    ifTrue: [ " try previous entry "
        sibling := super at: (nodeIndex - self entrySize).
        numSiblingElements := sibling numElements.
        numSiblingElements < (node mergeThreshold + 1)
            ifFalse: [ " can take entries from previous sibling "
                fromPreviousSibling := true.
                siblingLoss := (numSiblingElements - node mergeThreshold) quo: 2.
                siblingLoss == 0 ifTrue: [ siblingLoss := 1 ].
                copyStart := (numSiblingElements - siblingLoss) * eSize + 1.
                copyEnd := numSiblingElements * eSize.
                numToMove := (node numElements) * eSize . 
                targetStart := 1
            ]
    ].

" if cannot take from previous or next sibling "
fromPreviousSibling | fromNextSibling
    ifFalse: [ ^ false ].

" copy from sibling and insert into node "
node _insertAt: targetStart 
     from: sibling fromStart: copyStart fromEnd: copyEnd
     numToMoveDown: numToMove .

node numElements: (node numElements + siblingLoss).

fromNextSibling
    ifTrue: [ 
        " shift next sibling entries "
        sibling _deleteNoShrinkFrom: copyStart to: copyEnd .

        sibling numElements: (numSiblingElements - siblingLoss).

        " update the receiver's key for this entry "
        lastChildKey := node at: (node _lastKeyIndex).
        self primitiveAt: (nodeIndex + 1) put: lastChildKey.
        self _insertEncryptionFor: lastChildKey value: node startingAt: (nodeIndex + 2).

        node _updateLastValue
    ].

fromPreviousSibling
    ifTrue: [
        sibling numElements: (numSiblingElements - siblingLoss).

        " update the receiver's key for previous entry "
        siblingLastIdx := sibling _lastKeyIndex .
        lastChildKey := sibling at: siblingLastIdx .
        self  primitiveAt: (nodeIndex + 1 - eSize) put: (sibling at: siblingLastIdx ).
        self _insertEncryptionFor: lastChildKey value: sibling startingAt: (nodeIndex + 2 - eSize).

        sibling _updateLastValue .

       " nil out sibling entries at end "
        sibling _deleteNoShrinkFrom: copyStart to: copyEnd .
    ].

^ true
%

category: 'Comparison Operators'
method: BtreeInteriorNode
_compareKey: aKey value: aValue lessThanEntryAt: index
                                                                                                           
"Performs a < comparison between aKey and the entry whose key is at the given
 index.  If the keys are equal, then use the OOP of the value for the
 comparison.  The default implementation uses no encryption."
                                                                                                           
" first compare the keys "
( aKey _idxForSortLessThan: (super at: index) )
    ifTrue: [ ^ true ].
                                                                                                           
^ false
%

category: 'Comparison Operators'
method: BtreeInteriorNode
_compareKey: aKey value: aValue greaterThanEntryAt: index
                                                                                                           
"Performs a > comparison between aKey and the entry whose key is at the given
 index.  If the keys are equal, then use the OOP of the value for the
 comparison.  The default implementation uses no encryption."
                                                                                                           
" first compare the keys "
( aKey _idxForSortGreaterThan: (super at: index) )
    ifTrue: [ ^ true ].
                                                                                                           
^ false
%

category: 'Comparison Operators'
method: BtreeInteriorNode
_compareValueOop: aValue greaterThanEntryValueOopAt: index 

""

^ aValue asOop > (super at: index - 1) lastValue asOop
%

category: 'Comparison Operators'
method: BtreeInteriorNode
_compareValueOop: aValue lessThanOrEqualToEntryValueOopAt: index 

""

^ aValue asOop <= (super at: index - 1) lastValue asOop
%

category: 'Comparison Operators'
method: BtreeInteriorNode
_compareKey: aKey value: aValue equalToEntryAt: index 

"Performs a = comparison between aKey and the entry whose key is at the given
 index.  If the keys are equal, use the OOP of the values as the next basis for
 comparison.  For interior nodes, the value is the last value stored in the
 last leaf node.  The default implementation uses no encryption."

^ (aKey _idxForSortEqualTo: (super at: index)) _and:
  [ aValue asOop = (super at: index - 1) lastValue asOop ]
%

category: 'Comparison Operators'
method: BtreeInteriorNode
_compareKey: aKey value: aValue lessThanOrEqualToEntryAt: index

"Performs a <= comparison between aKey and the entry whose key is at the given
 index.  If the keys are equal, then use the OOP of the value for the
 comparison.  For interior nodes, the value is the last value stored in the
 last leaf node.  The default implementation uses no encryption."

" first compare the keys "
( aKey _idxForSortLessThan: (super at: index) )
    ifTrue: [ ^ true ].

" if the keys are equal, use the OOP of the value "
(aKey _idxForSortEqualTo: (super at: index))
    ifTrue: [ ^ aValue asOop <= (super at: index - 1) lastValue asOop ].

^ false
%

category: 'Searching'
method: BtreeInteriorNode
_findAllValuesForKey: aKey into: aCollection using: aComparison

"Finds all the values for the given key, placing them in the collection.  For
 interior nodes, must first find the appropriate child node.  While child nodes
 find additional values for the key, continue with the next child."

| index eSize prevSize maxIndex |
" find first child node that might contain the key "
index := self _findCoveringIndexForKey: aKey totalOrder: false.
index == 0 ifTrue: [ ^ self ].

prevSize := aCollection size.
(super at: index) _findAllValuesForKey: aKey into: aCollection using: aComparison.

eSize := self entrySize.
maxIndex := self _lastIndex.
index := index + eSize.

" now scan child nodes until no values are added to the collection "
[ (index < maxIndex) _and:
[ prevSize < aCollection size ] ] whileTrue: [
    prevSize := aCollection size.
    (super at: index) _scanForAllValuesForKey: aKey into: aCollection using: aComparison.
    index := index + eSize
]
%

category: 'Searching'
method: BtreeInteriorNode
_findAllValuesForIdenticalKey: aKey into: aCollection using: aComparison

"Finds all the values for the given identical key, placing them in the
 collection.  For interior nodes, must first find the appropriate child
 node.  Return whether the last key was equal or not (so the caller
 knows it must check the next node)."

| index eSize maxIndex |
" find first child node that might contain the key "
index := self _findCoveringIndexForKey: aKey totalOrder: false.
index == 0 ifTrue: [ ^ false ].

((super at: index) _findAllValuesForIdenticalKey: aKey into: aCollection using: aComparison)
  ifFalse: [ ^ false ].

eSize := self entrySize.
maxIndex := self _lastIndex.
index := index + eSize.

" now scan child nodes until no values are added to the collection "
[ (index < maxIndex) _and:
[ (super at: index) _findAllValuesForIdenticalKey: aKey into: aCollection using: aComparison ] ]
  whileTrue: [ index := index + eSize ].
^ index >= maxIndex
%

category: 'Searching Support'
method: BtreeInteriorNode
_findAllValuesGreaterThanKey: aKey into: anArray using: aComparison

"Puts into anArray the index and the leaf node of the entry that is greater
 than the given key, plus the index and interior node of the path taken to
 reach the leaf.  Returns true if a value was found, false otherwise.  Do this
 by invoking the same operation on the appropriate child node."

| index eSize lastIndex |
lastIndex := self _lastKeyIndex.
lastIndex == 0
    ifTrue: [ ^ false ].

" find index where key would be inserted "
index := self _findCoveringIndexForKey: aKey totalOrder: false.
index == 0
    ifTrue: [ ^ false ].

"scan across all entries that sort ="
eSize := self entrySize.
[ self _compare: BtreeComparisonForSort key: aKey equalToEntryAt: index + 1 ] whileTrue: [
    index := index + eSize.
    index > lastIndex ifTrue: [ ^ false ]
].

" see if this child has any greater than the key "
((super at: index) _findAllValuesGreaterThanKey: aKey into: anArray using: aComparison)
    ifFalse: [ ^ false ].

anArray addLast: index.
anArray addLast: self.
^ true
%

category: 'Searching Support'
method: BtreeInteriorNode
_findAllValuesLessThanKey: aKey into: anArray using: aComparison

"Puts into anArray the index and the leaf node of the entry that is less than
 (or less than and equal to if aBoolean is true) the given key, plus the index
 and interior node of the path taken to reach the leaf.  Returns true if a
 value was found, false otherwise.  Does this by invoking the same operation
 on the appropriate child node."

| index eSize lastIndex |
lastIndex := self _lastIndex.
lastIndex == 0
  ifTrue: [ ^ false ].

" first see if all entries are > aKey "
(self _compare: BtreeComparisonForSort key: aKey lessThanOrEqualToEntryAt: self _lastKeyIndex)
  ifFalse: [
    "bugfix 33805 - verify that aKey is > last entry"
    ((aKey == nil) _or: [self _compare: aComparison key: aKey greaterThanEntryAt: self _lastKeyIndex])
      ifTrue: [
        self _putLastIndexOfLastChildInto: anArray.
        ^ true
      ].
  ].
" find index where key would be inserted "
index := self _findCoveringIndexForKey: aKey totalOrder: false.
index == 0
    ifTrue: [ ^ false ].

eSize := self entrySize.

" see if this child has any less than the key "
((super at: index) _findAllValuesLessThanKey: aKey into: anArray using: aComparison)
  ifFalse: [ " this child did not find any "
    " if it is not the first child "
    index > 1
      ifTrue: [
        " then the entry is the rightmost leaf entry of the previous child "
        index := index - eSize.
        (super at: index) _putLastIndexOfLastChildInto: anArray
      ]
      ifFalse: [ ^ false ]
  ].
anArray addLast: index.
anArray addLast: self.
^ true
%

category: 'Searching Support'
method: BtreeInteriorNode
_findAllValuesLessThanKey: aKey andEquals: aBoolean into: anArray using: aComparison

"Puts into anArray the index and the leaf node of the entry that is less than
 (or less than and equal to if aBoolean is true) the given key, plus the index
 and interior node of the path taken to reach the leaf.  Returns true if a
 value was found, false otherwise.  Does this by invoking the same operation
 on the appropriate child node."

| index eSize lastIndex |
lastIndex := self _lastIndex.
lastIndex == 0
  ifTrue: [ ^ false ].

aBoolean 
  ifFalse: [ ^self  _findAllValuesLessThanKey: aKey into: anArray using: aComparison ].

" first see if all entries are > aKey "
(self _compare: BtreeComparisonForSort key: aKey lessThanOrEqualToEntryAt: self _lastKeyIndex)
  ifFalse: [
    "bugfix 33805 - verify that aKey is > last entry"
    ((aKey == nil) _or: [self _compare: aComparison key: aKey greaterThanEntryAt: self _lastKeyIndex])
      ifTrue: [
        self _putLastIndexOfLastChildInto: anArray.
        ^ true
      ].
  ].
"find index of first entry >= aKey"
index := self _findCoveringIndexForKey: aKey totalOrder: false.
index == 0
    ifTrue: [ ^ false ].

"scan across all of the values that sort ="
eSize := self entrySize.
(self _compare:  BtreeComparisonForSort key: aKey equalToEntryAt: index + 1)
  ifTrue: 
    [ [ index < lastIndex _and:
        [ self _compare:  BtreeComparisonForSort key: aKey equalToEntryAt: index + 1 ] ]
          whileTrue: [ index := index + eSize ].
    " if we scanned past the end, subtract an entry "
    index > lastIndex ifTrue: [ index := index - eSize ]
    ].

" see if child has any entries less than or equal to aKey "
((super at: index) _findAllValuesLessThanKey: aKey andEquals: true into: anArray using: BtreeComparisonForSort)
  ifFalse: [ " this child did not find any "
    " if it is not the first child "
    index > 1
      ifTrue: [
        " then the entry is the rightmost leaf entry of the previous child "
        index := index - eSize.
        (super at: index) _putLastIndexOfLastChildInto: anArray
      ]
      ifFalse: [ ^ false ]
  ].
anArray addLast: index.
anArray addLast: self.
^ true
%

category: 'Searching'
method: BtreeInteriorNode
_findFirstValueForKey: aKey using: aComparison

"Returns the first value associated with the given key.  For interior nodes,
 find the appropriate child node and ask it to find the first value."

| index |
" find first child node that might contain the key "
index := self _findCoveringIndexForKey: aKey totalOrder: false.
index == 0 ifTrue: [ ^ #_incompletePathTraversal ].

^ (super at: index) _findFirstValueForKey: aKey using: aComparison
%

category: 'Updating'
method: BtreeInteriorNode
_insertEncryptionFor: aKey value: node startingAt: insertionIndex

"Copies the encryption present in the child node to the receiver."

| nodeIndex eSize |

" if it exists, insert the encryption (copy from child node) "
eSize := node _encryptionSize.
eSize > 0
    ifTrue: [
        nodeIndex := node _lastKeyIndex.
        node
            copyFrom: (nodeIndex + 1)
            to: (nodeIndex + eSize)
            into: self
            startingAt: insertionIndex
    ]
%

category: 'Removing'
method: BtreeInteriorNode
_mergeFromNeighborForNodeAt: nodeIndex

"The node at the given index has too few entries and neighboring siblings had
 too few entries to take from, so merge with one of its neighboring siblings.
 First tries the next (right) sibling, then the previous (left) sibling."

| node sibling eSize myESize numSiblingElements copyStart lastChildKey myLastIndex |

node := super at: nodeIndex.
eSize := node entrySize.
myESize := self entrySize.

" check if the node is the last entry "
nodeIndex == (self _lastKeyIndex - 1)
    ifTrue: [ " this is the last entry, try the previous sibling "
        nodeIndex > 1
            ifTrue: [ " keep sibling, throw away node "
                sibling := super at: (nodeIndex - myESize).
                numSiblingElements := sibling numElements.

                " move node's entries onto end of sibling "
                sibling _insertAt: (sibling _lastIndex + 1)
                   from: node fromStart: 1 fromEnd: (node numElements * eSize)
                   numToMoveDown: 0 . 

                sibling numElements: (node numElements + numSiblingElements).

                " update the key for the sibling's entry in the receiver "
                lastChildKey := sibling at: (sibling _lastKeyIndex).
                self primitiveAt: (nodeIndex - myESize + 1) put: lastChildKey.
                self
                    _insertEncryptionFor: lastChildKey
                    value: sibling
                    startingAt: (nodeIndex - myESize + 2).

                sibling _updateLastValue
            ]
            ifFalse:[ self _error: #rtErrBtreeInvalidMerge.
                      self _uncontinuableError ]
    ]
    " this is not the last entry, get the next sibling "
    ifFalse: [ " keep node, throw away sibling "
        sibling := super at: (nodeIndex + myESize).
        numSiblingElements := sibling numElements.

        "copy from sibling onto end of node "
        node _insertAt: (node _lastIndex + 1) 
             from: sibling fromStart: 1 fromEnd: (numSiblingElements * eSize)
             numToMoveDown: 0 .

        copyStart := nodeIndex + myESize + myESize.
        " if sibling was not the last entry in the receiver "
        myLastIndex := self _lastIndex .
        copyStart < myLastIndex
            ifTrue: [
                " move entries in receiver to the left "
                self _deleteNoShrinkFrom: (nodeIndex + myESize) 
			to: copyStart - 1 
            ].

        node numElements: (node numElements + numSiblingElements).

        " update the key for the the node's entry in the receiver "
        lastChildKey := node at: (node _lastKeyIndex).
        self primitiveAt: (nodeIndex + 1) put: lastChildKey.
        self
            _insertEncryptionFor: lastChildKey
            value: node
            startingAt: (nodeIndex + 2).

        node _updateLastValue
    ].

numElements := numElements - 1.

" nil out the last entry of the receiver "
myLastIndex := self _lastIndex .
self _deleteNoShrinkFrom: myLastIndex + 1 to: myLastIndex + myESize.

%

category: 'Searching Support'
method: BtreeInteriorNode
_putFirstIndexOfFirstChildInto: anArray

"Ask the first child in the receiver to
 putFirstIndexOfFirstChildInto: anArray."

(super at: 1) _putFirstIndexOfFirstChildInto: anArray.
anArray addLast: 1.
anArray addLast: self
%

category: 'Searching Support'
method: BtreeInteriorNode
_putFirstIndexOfFirstChildInto: anArray ifGreaterThanOrEqualTo: aKey using: aComparison

"Ask the first child in the receiver to putFirstIndexOfFirstChildInto:ifGreaterThanOrEqualTo:.

 Because of NANs we need to validate the assumption of the caller 
 (i.e., that all entries are greater than aKey). See bug33805."

"make sure that key is less than or equal to first entry"
(aKey == nil 
   _or: [ self _compare: aComparison key: aKey lessThanOrEqualToEntryAt: 2 ])
  ifTrue: [
    (super at: 1) _putFirstIndexOfFirstChildInto: anArray ifGreaterThanOrEqualTo: aKey using: aComparison.
    anArray addLast: 1.
  ]
  ifFalse: [
    anArray addLast: 0.
  ].
anArray addLast: self.
%

category: 'Searching Support'
method: BtreeInteriorNode
_putLastIndexOfLastChildInto: anArray

"Asks the last child in the receiver to
 putLastIndexOfLastChildInto: anArray ."

(super at: self _lastKeyIndex - 1) _putLastIndexOfLastChildInto: anArray.
anArray addLast: self _lastKeyIndex - 1.
anArray addLast: self
%

category: 'Searching'
method: BtreeInteriorNode
_scanForAllValuesForKey: aKey into: aCollection using: aComparison

"Places all values associated with the given key into aCollection.
 Start with the first entry, and iterate through consecutive entries
 until the first entry that does not add to the collection."

| index eSize prevSize maxIndex |
prevSize := aCollection size.
(super at: 1) _findAllValuesForKey: aKey into: aCollection using: aComparison.

eSize := self entrySize.
maxIndex := self _lastIndex.
index := 1 + eSize.

[ (index < maxIndex) _and:
[ prevSize < aCollection size ] ] whileTrue: [
    prevSize := aCollection size.
    (super at: index) _scanForAllValuesForKey: aKey into: aCollection using: aComparison.
    index := index + eSize
]
%

category: 'Updating'
method: BtreeInteriorNode
_splitUsingKey: aKey value: aNode

"Creates a new sibling node, moving half of the receiver's entries into
 the sibling.  Then insert a new entry with the given key/node into either
 the receiver or sibling.  Returns the new sibling."

| insertionIndex newSibling lastKeyIndex lastKey lastVal node |
newSibling := self _createSibling.

" get the last key in the receiver "
lastKeyIndex := self _lastKeyIndex.
lastKey := super at: lastKeyIndex.
" get the last value in the new entry's node (this is what insertion is based upon) "
lastVal := aNode lastValue.

" determine if new entry goes in the first half or second half "
(self _compareKey: aKey value: lastVal lessThanOrEqualToEntryAt: lastKeyIndex)
  ifTrue: [  " adding to the first half "
    insertionIndex := self _binarySearchCoveringKey: aKey value: lastVal.

    " HR6190 "
    (node := self _at: insertionIndex) ~~ nil
      ifTrue: [
        (node _compareKey: aKey value: lastVal lessThanOrEqualToEntryAt: 2)
          ifFalse: [ insertionIndex := insertionIndex + self entrySize ].
      ].

    self _insertKey: aKey value: aNode atIndex: insertionIndex
  ]
  ifFalse: [ " adding to the second half "
    insertionIndex := newSibling _binarySearchCoveringKey: aKey value: lastVal.

    " HR6190 "
    (insertionIndex <= newSibling size) ifTrue:[                "fix 31179"
      (node := newSibling _at: insertionIndex) ~~ nil
        ifTrue: [
          (node _compareKey: aKey value: lastVal lessThanOrEqualToEntryAt: 2)
            ifFalse: [ insertionIndex := insertionIndex + self entrySize ].
        ].
    ].
    newSibling _insertKey: aKey value: aNode atIndex: insertionIndex
  ].

^ newSibling
%

category: 'Updating'
method: BtreeInteriorNode
_updateLastValue

"Update the last value of the interior node by asking the last entry what its
 last value is.  Do not change the value unless there has been a change."

| nodeLastValue |

nodeLastValue := (super at: (self _lastKeyIndex - 1)) lastValue.
(nodeLastValue == lastValue)
    ifFalse: [ lastValue := nodeLastValue ]
%

category: 'Updating'
method: BtreeInteriorNode
btreeAt: aKey put: aValue

""

^ self at: aKey put: aValue
%

category: 'Updating'
method: BtreeInteriorNode
at: aKey put: aValue

"Adds the key/value pair to one of the leaf nodes.  If a split occurs, returns
 the new sibling, otherwise returns the receiver."

| index node returnNode newSibling lastKeyIndex lastChildKey lastChildKeyIndex |

index := self _binarySearchCoveringKey: aKey value: aValue.

lastKeyIndex := self _lastKeyIndex.
" if insertion would occur past the last entry, use the last entry "
index > lastKeyIndex
    ifTrue: [ index := lastKeyIndex - 1 ].

node := super at: index.
returnNode := node at: aKey put: aValue.

" see if last key in child node has changed "
lastChildKeyIndex := node _lastKeyIndex.
lastChildKey := node _at: lastChildKeyIndex.
(lastChildKey == (super at: (index + 1)))
    ifFalse: [ " update key and encryption in the receiver "
        self primitiveAt: (index + 1) put: lastChildKey.
        self _insertEncryptionFor: lastChildKey value: node startingAt: (index + 2)
    ].

" see if a split occurred "
returnNode == node
    ifTrue: [
        " if inserted into the last node, update the last value "
        (index == (lastKeyIndex - 1))
            ifTrue: [ self _updateLastValue ].
    ]
    ifFalse: [ " returnNode is the second half of the split "

        " determine if a split is needed in the receiver "
        numElements == self class maxNumberOfElements
            ifTrue: [
                newSibling :=
                    self _splitUsingKey: (returnNode _at: returnNode _lastKeyIndex)
                        value: returnNode.

                " update the last value of the new sibling and the receiver "
                newSibling lastValue:
                    (newSibling at: (newSibling _lastKeyIndex - 1)) lastValue.
                lastValue := (super at: (self _lastKeyIndex - 1)) lastValue.

                ^ newSibling
            ]
            ifFalse: [ " entry will fit in the receiver "

                " if child node was the last node "
                index == (lastKeyIndex - 1)
                    ifTrue: [ lastValue := returnNode lastValue ].

                self
                    _insertKey: (returnNode _at: returnNode _lastKeyIndex)
                    value: returnNode
                    atIndex: (index + returnNode entrySize)
            ]
    ]
%

category: 'Accessing'
method: BtreeInteriorNode
lastValue

"Returns the value of the instance variable 'lastValue'."

^ lastValue
%

category: 'Updating'
method: BtreeInteriorNode
lastValue: newValue

"Modifies the value of the instance variable 'lastValue'."

lastValue := newValue
%

category: 'Constants'
method: BtreeInteriorNode
mergeThreshold

"Returns the threshold such that if an entry has less than this number, it must
 have elements added from its sibling or be merged with its sibling.  This
 value is usually the same as the result of the message
 'self maxNumberOfElements quo: 2'."

^ 500
%

category: 'Printing'
method: BtreeInteriorNode
printOn: aStream withIndentationLevel: anInteger

"Puts a displayable representation of the keys of the receiver on the given
 stream, indenting the given number of spaces."

anInteger timesRepeat: [ aStream nextPut: Character space ].
aStream nextPutAll: '[ ';
    nextPutAll: lastValue asString;
    nextPutAll: ' ]';

    "nextPutAll: '  { '; nextPutAll: self _alias asString; nextPutAll: ' }';"
    nextPut: Character lf.

1 to: (numElements * self entrySize) by: self entrySize do: [ :i |
    anInteger timesRepeat: [ aStream nextPut: Character space ].
    aStream nextPutAll: (super at: (i + 1)) asString; nextPut: Character lf.
    (super at: i) printOn: aStream withIndentationLevel: (anInteger + 4)
]
%


category: 'Printing'
method: BtreeInteriorNode
printOn: aStream withIndentationLevel: anInteger includingValues: aBoolean

"Puts a displayable representation of the keys of the receiver on the given
 stream, indenting the given number of spaces."

aBoolean
    ifTrue: [
        anInteger timesRepeat: [ aStream nextPut: Character space ].
        aStream nextPutAll: '[ ';
            nextPutAll: lastValue asString;
            nextPutAll: ' ]';
            "nextPutAll: '  { '; nextPutAll: self _alias asString; nextPutAll: ' }';"
            nextPut: Character lf.
    ].

1 to: (numElements * self entrySize) by: self entrySize do: [ :i |
    anInteger timesRepeat: [ aStream nextPut: Character space ].
    aStream nextPutAll: (super at: (i + 1)) asString; nextPut: Character lf.
    (super at: i) printOn: aStream withIndentationLevel: (anInteger + 4) includingValues: aBoolean
]
%

category: 'Removing'
method: BtreeInteriorNode
removeKey: aKey value: aValue

"Removes the key/value pair.  Find the appropriate child node and invoke the
 remove operation on it.  If the remove occurred, check if a merge or coalesce
 is necessary."

| nodeIndex node lastChildKeyIndex lastChildKey |
(self _compareLastKeyLessThanOrEqualToKey: aKey value: aValue)
    ifFalse: [ ^ false ].

nodeIndex := self _binarySearchCoveringKey: aKey value: aValue.
node := super at: nodeIndex.

(node removeKey: aKey value: aValue)
    ifTrue: [ " remove occurred "

        " see if last key in child node has changed "
        lastChildKeyIndex := node _lastKeyIndex.
        lastChildKey := node at: lastChildKeyIndex.
        (lastChildKey == (super at: (nodeIndex + 1)))
            ifFalse: [ " update key and encryption in the receiver "
                self primitiveAt: (nodeIndex + 1) put: lastChildKey.
                self
                    _insertEncryptionFor: lastChildKey
                    value: node
                    startingAt: (nodeIndex + 2)
            ].

        " see if the node's number of entries is below the threshold "
        node numElements < node mergeThreshold
            ifTrue: [ " try to steal entries from a neighboring sibling "
                (self _attemptTakingFromNeighborForNodeAt: nodeIndex)
                    ifFalse: [
                        " could not take from neighbor, merge with a neighbor instead "
                        self _mergeFromNeighborForNodeAt: nodeIndex
                    ]
            ].

        self _updateLastValue.

        ^ true
    ]
    ifFalse: [ ^ false ]
%

category: 'Printing'
method: BtreeInteriorNode
stringRep

""

| aStream |

aStream := PrintStream printingOn: String new.

1 to: (numElements * self entrySize) by: self entrySize do: [ :i |
    aStream nextPutAll: (super at: (i + 1)) asString; nextPut: Character space
].
aStream nextPut: Character lf.
^ aStream contents
%

category: 'Enumerating'
method: BtreeInteriorNode
_preOrderDo: aBlock

"Execute the one-argument block with the receiver, then
 invoke the same message on the child nodes."

aBlock value: self.

1 to: (numElements * self entrySize) by: self entrySize do: [ :i |
    (super at: i) _preOrderDo: aBlock
]
%

category: 'Enumerating'
method: BtreeInteriorNode
_leafKeysAndValuesDo: aBlock

"Execute the two-argument block for each child node."

1 to: (numElements * self entrySize) by: self entrySize do: [ :i |
    (super at: i) _leafKeysAndValuesDo: aBlock
]
%

category: 'Private'
method: BtreeInteriorNode
numElementsInStack: array
level: level
endNode: endNode
endIndex: endIndex
into: result

"Puts the count in the first slot in the argument array and returns whether we
 reached the end."

| count finished |
" save current count "
count := result at: 1.

" for each child node starting at the node on the stack "
(array at: level) to: (numElements * self entrySize) by: self entrySize do: [ :i |
    " reset the count to zero "
    result at: 1 put: 0.

    " recursive call to child node "
    finished := (super at: i) numElementsInStack: array 
        level: level - 2 
        endNode: endNode 
        endIndex: endIndex 
        into: result.

    " accumulate the count "
    count := count + (result at: 1).
    " if reached the end node, put the count in the result Array "
    finished
        ifTrue: [
            result at: 1 put: count.
            ^ true
        ]
].
" put the count in the result Array "
result at: 1 put: count.
^ false
%

! deleted _lastValuesNotIdenticalTo:into: v2.0

category: 'Sorting Support'
method: BtreeInteriorNode
_putAscendingValuesInto: array startingAt: offset

"Returns the number of elements."

| j |
j := offset.
1 to: (numElements * self entrySize) by: self entrySize do: [ :i |
  j := j + ((super at: i) _putAscendingValuesInto: array startingAt: j)
].
^ j - offset
%

category: 'Sorting Support'
method: BtreeInteriorNode
_putDescendingValuesInto: array startingAt: offset

"Returns the number of elements."

| j |
j := offset.
1 to: (numElements * self entrySize) by: self entrySize do: [ :i |
  j := j - ((super at: i) _putDescendingValuesInto: array startingAt: j)
].
^ offset - j
%

category: 'Accessing'
method: BtreeInteriorNode
_totalElementsIn: stack endNode: endNode endIndex: endIndex into: array

"Determine the total number of entries, checking the given stack and
 end node from a stream object.  The given Array contains the current
 sum of elements in the first slot, and the second slots indicates if
 the end was hit."

| i start |
(i := stack indexOf: self) > 0
  ifTrue: [ start := stack at: i - 1 ]
  ifFalse: [ start := 1 ].

start to: self _lastIndex by: self entrySize do: [ :j |
  (super at: j) _totalElementsIn: stack
    endNode: endNode
    endIndex: endIndex
    into: array.
  (array at: 2) ifTrue: [ ^ self ]
].
%

category: 'Accessing'
method: BtreeInteriorNode
totalElements

"Return the total number of elements in the leaf nodes."

| sum |

sum := 0.
1 to: (numElements * self entrySize) by: self entrySize do: [ :i |
    sum := sum + (super at: i) totalElements
].
^ sum
%

category: 'Accessing'
method: BtreeInteriorNode
changeToSegment: aSegment

"Assign the receiver and its leaf nodes the given segment."

"Prints a string representation of the keys of the receiver on
 the stream, indenting the given number of spaces."

self assignToSegment: aSegment.

1 to: (numElements * self entrySize) by: self entrySize do: [ :i |
    (super at: i) changeToSegment: aSegment
]
%

category: 'RC Removing'
method: BtreeInteriorNode
_attemptTakingFromNeighborForNodeAt: nodeIndex for: anIndexObj logging: aBoolean

"The node at the given index has too few entries, so try to take entries from
 one of its neighboring siblings.  First try the next (right) sibling, then
 the previous (left) sibling."

| node sibling eSize siblingLoss numSiblingElements copyStart copyEnd
fromNextSibling fromPreviousSibling targetStart lastChildKey 
siblingLastIdx numToMove |

node := super at: nodeIndex.
fromNextSibling := false.  " is true if taken from next sibling "
fromPreviousSibling := false.  " is true if taken from previous sibling "

eSize := node entrySize.
" check if the node is the last entry "
nodeIndex == (self _lastKeyIndex - 1)
    ifFalse: [ " this is not the last entry, check the next sibling "
        sibling := super at: (nodeIndex + self entrySize).
        numSiblingElements := sibling numElements.
        numSiblingElements < (node mergeThreshold + 1)
            ifFalse: [ " can take entries from next sibling "
                fromNextSibling := true.
                siblingLoss := (numSiblingElements - node mergeThreshold) quo: 2.
                siblingLoss == 0 ifTrue: [ siblingLoss := 1 ].
                copyStart := 1.
                copyEnd := siblingLoss * eSize.
                numToMove := 0 .
                targetStart := node _lastIndex + 1
            ]
    ].

" if could not get entries from next sibling and this is not the first entry "
(fromNextSibling not _and: [ nodeIndex > 1 ] )
    ifTrue: [ " try previous entry "
        sibling := super at: (nodeIndex - self entrySize).
        numSiblingElements := sibling numElements.
        numSiblingElements < (node mergeThreshold + 1)
            ifFalse: [ " can take entries from previous sibling "
                fromPreviousSibling := true.
                siblingLoss := (numSiblingElements - node mergeThreshold) quo: 2.
                siblingLoss == 0 ifTrue: [ siblingLoss := 1 ].
                copyStart := (numSiblingElements - siblingLoss) * eSize + 1.
                copyEnd := numSiblingElements * eSize.
                numToMove := (node numElements) * eSize . 
                targetStart := 1
            ]
    ].

" if cannot take from previous or next sibling "
fromPreviousSibling | fromNextSibling
    ifFalse: [ ^ false ].

" copy from sibling and insert into node "
node _insertAt: targetStart 
     from: sibling fromStart: copyStart fromEnd: copyEnd
     numToMoveDown: numToMove .

node numElements: (node numElements + siblingLoss).

fromNextSibling
    ifTrue: [ 
        " shift next sibling entries "
        sibling _deleteNoShrinkFrom: copyStart to: copyEnd .

        sibling numElements: (numSiblingElements - siblingLoss).

        " update the receiver's key for this entry "
        lastChildKey := node at: (node _lastKeyIndex).
        self primitiveAt: (nodeIndex + 1) put: lastChildKey.
        self _insertEncryptionFor: lastChildKey value: node startingAt: (nodeIndex + 2).

        node _updateLastValue
    ].

fromPreviousSibling
    ifTrue: [
        sibling numElements: (numSiblingElements - siblingLoss).

        " update the receiver's key for previous entry "
        siblingLastIdx := sibling _lastKeyIndex .
        lastChildKey := sibling at: siblingLastIdx .
        self  primitiveAt: (nodeIndex + 1 - eSize) put: (sibling at: siblingLastIdx ).
        self _insertEncryptionFor: lastChildKey value: sibling startingAt: (nodeIndex + 2 - eSize).

        sibling _updateLastValue .

       " nil out sibling entries at end "
        sibling _deleteNoShrinkFrom: copyStart to: copyEnd .
    ].

System _addRootObjectToRcReadSet: node.
System _addRootObjectToRcReadSet: sibling.
System redoLog addConflictObject: node for: anIndexObj.
System redoLog addConflictObject: sibling for: anIndexObj.
^ true
%
category: 'RC Removing'
method: BtreeInteriorNode
_mergeFromNeighborForNodeAt: nodeIndex for: anIndexObj logging: aBoolean

"The node at the given index has too few entries and neighboring siblings had
 too few entries to take from, so merge with one of its neighboring siblings.
 First tries the next (right) sibling, then the previous (left) sibling."

| node sibling eSize myESize numSiblingElements copyStart lastChildKey myLastIndex |

node := super at: nodeIndex.
eSize := node entrySize.
myESize := self entrySize.

" check if the node is the last entry "
nodeIndex == (self _lastKeyIndex - 1)
    ifTrue: [ " this is the last entry, try the previous sibling "
        nodeIndex > 1
            ifTrue: [ " keep sibling, throw away node "
                sibling := super at: (nodeIndex - myESize).
                numSiblingElements := sibling numElements.

                " move node's entries onto end of sibling "
                sibling _insertAt: (sibling _lastIndex + 1)
                   from: node fromStart: 1 fromEnd: (node numElements * eSize)
                   numToMoveDown: 0 . 

                sibling numElements: (node numElements + numSiblingElements).

                " update the key for the sibling's entry in the receiver "
                lastChildKey := sibling at: (sibling _lastKeyIndex).
                self primitiveAt: (nodeIndex - myESize + 1) put: lastChildKey.
                self
                    _insertEncryptionFor: lastChildKey
                    value: sibling
                    startingAt: (nodeIndex - myESize + 2).

                sibling _updateLastValue
            ]
            ifFalse:[ self _error: #rtErrBtreeInvalidMerge.
                      self _uncontinuableError ]
    ]
    " this is not the last entry, get the next sibling "
    ifFalse: [ " keep node, throw away sibling "
        sibling := super at: (nodeIndex + myESize).
        numSiblingElements := sibling numElements.

        "copy from sibling onto end of node "
        node _insertAt: (node _lastIndex + 1) 
             from: sibling fromStart: 1 fromEnd: (numSiblingElements * eSize)
             numToMoveDown: 0 .

        copyStart := nodeIndex + myESize + myESize.
        " if sibling was not the last entry in the receiver "
        myLastIndex := self _lastIndex .
        copyStart < myLastIndex
            ifTrue: [
                " move entries in receiver to the left "
                self _deleteNoShrinkFrom: (nodeIndex + myESize) 
			to: copyStart - 1 
            ].

        node numElements: (node numElements + numSiblingElements).

        " update the key for the the node's entry in the receiver "
        lastChildKey := node at: (node _lastKeyIndex).
        self primitiveAt: (nodeIndex + 1) put: lastChildKey.
        self
            _insertEncryptionFor: lastChildKey
            value: node
            startingAt: (nodeIndex + 2).

        node _updateLastValue
    ].

System _addRootObjectToRcReadSet: node.
System _addRootObjectToRcReadSet: sibling.
System redoLog addConflictObject: node for: anIndexObj.
System redoLog addConflictObject: sibling for: anIndexObj.

numElements := numElements - 1.

" nil out the last entry of the receiver "
myLastIndex := self _lastIndex .
self _deleteNoShrinkFrom: myLastIndex + 1 to: myLastIndex + myESize.

%
category: 'RC Removing'
method: BtreeInteriorNode
_removeKey: aKey value: aValue for: anIndexObj logging: aBoolean
"Removes the key/value pair.  Find the appropriate child node and invoke the
 remove operation on it.  If the remove occurred, check if a merge or coalesce
 is necessary."

| nodeIndex node lastChildKeyIndex lastChildKey |
(self _compareLastKeyLessThanOrEqualToKey: aKey value: aValue)
    ifFalse: [ ^ false ].

nodeIndex := self _binarySearchCoveringKey: aKey value: aValue.
node := super at: nodeIndex.

(node removeKey: aKey value: aValue for: anIndexObj logging: aBoolean)
    ifTrue: [ " remove occurred "

        " see if last key in child node has changed "
        lastChildKeyIndex := node _lastKeyIndex.
        lastChildKey := node at: lastChildKeyIndex.
        (lastChildKey == (super at: (nodeIndex + 1)))
            ifFalse: [ " update key and encryption in the receiver "
                self primitiveAt: (nodeIndex + 1) put: lastChildKey.
                self
                    _insertEncryptionFor: lastChildKey
                    value: node
                    startingAt: (nodeIndex + 2)
            ].

        " see if the node's number of entries is below the threshold "
        node numElements < node mergeThreshold
            ifTrue: [ " try to steal entries from a neighboring sibling "
                (self _attemptTakingFromNeighborForNodeAt: nodeIndex for: anIndexObj logging: aBoolean)
                    ifFalse: [
                        " could not take from neighbor, merge with a neighbor instead "
                        self _mergeFromNeighborForNodeAt: nodeIndex for: anIndexObj logging: aBoolean
                    ]
            ].

        self _updateLastValue.

        ^ true
    ]
    ifFalse: [ ^ false ]
%

category: 'RC Updating'
method: BtreeInteriorNode
_at: aKey put: aValue for: anIndexObj logging: aBoolean

"Adds the key/value pair to one of the leaf nodes.  If a split occurs, returns
 the new sibling, otherwise returns the receiver."

| index node returnNode newSibling lastKeyIndex lastChildKey lastChildKeyIndex |

index := self _binarySearchCoveringKey: aKey value: aValue.

lastKeyIndex := self _lastKeyIndex.
" if insertion would occur past the last entry, use the last entry "
index > lastKeyIndex
    ifTrue: [ index := lastKeyIndex - 1 ].

node := super at: index.
returnNode := node btreeAt: aKey put: aValue for: anIndexObj logging: aBoolean.

" see if last key in child node has changed "
lastChildKeyIndex := node _lastKeyIndex.
lastChildKey := node _at: lastChildKeyIndex.
(lastChildKey == (super at: (index + 1)))
    ifFalse: [ " update key and encryption in the receiver "
        self primitiveAt: (index + 1) put: lastChildKey.
        self _insertEncryptionFor: lastChildKey value: node startingAt: (index + 2)
    ].

" see if a split occurred "
returnNode == node
    ifTrue: [
        " if inserted into the last node, update the last value "
        (index == (lastKeyIndex - 1))
            ifTrue: [ self _updateLastValue ].
    ]
    ifFalse: [ " returnNode is the second half of the split "

        System _addRootObjectToRcReadSet: returnNode.
        System redoLog addConflictObject: returnNode for: anIndexObj.
        " determine if a split is needed in the receiver "
        numElements == self class maxNumberOfElements
            ifTrue: [
                newSibling :=
                    self _splitUsingKey: (returnNode _at: returnNode _lastKeyIndex)
                        value: returnNode.

                " update the last value of the new sibling and the receiver "
                newSibling lastValue:
                    (newSibling at: (newSibling _lastKeyIndex - 1)) lastValue.
                lastValue := (super at: (self _lastKeyIndex - 1)) lastValue.

                System _addRootObjectToRcReadSet: newSibling.
                System redoLog addConflictObject: newSibling for: anIndexObj.
                ^ newSibling
            ]
            ifFalse: [ " entry will fit in the receiver "

                " if child node was the last node "
                index == (lastKeyIndex - 1)
                    ifTrue: [ lastValue := returnNode lastValue ].

                self
                    _insertKey: (returnNode _at: returnNode _lastKeyIndex)
                    value: returnNode
                    atIndex: (index + returnNode entrySize)
            ]
    ]
%
