from typing import (
Self,
ClassVar,
)
from abc import (
ABC,
abstractmethod,
)
from collections.abc import Iterator
from bluebase.core import (
LayoutField,
LayoutMixin,
Value,
Domain,
dump,
assignment,
is_value_equal,
)
from bluebase.pf import (
PfPageId,
PfPageData,
PfPageView,
PfPageHeader,
PfPage,
)
from bluebase.rm import (
RmRecordId,
RmSlotId,
RmPageId,
)
from .bucket import (
IxBucketId,
)
from .type import (
IxSlotId,
IxNodeId,
IxNodeData,
)
__all__ = (
'IxPointer',
'IxLeafPointer',
'IxBranchPointer',
'IxNodeHeader',
'IxNode',
'IxLeafNode',
'IxBranchNode',
)
[docs]
class IxPointer(LayoutMixin, ABC):
"""
.. rubric:: Layout
.. list-table::
:header-rows: 1
* - Offset
- Size
- Name
- Format
* - 0
- 4
- `rid_pid`
- ``'<i'``
* - 4
- 4
- `rid_sid`
- ``'<i'``
* - 8
- 4
- `link`
- ``'<i'``
"""
NULL: ClassVar[PfPageId] = PfPageId(-1)
"""pointer가 아무것도 가리키지 않을 때 사용하는 상수."""
fields = (
LayoutField('rid_pid', 'i'),
LayoutField('rid_sid', 'i'),
LayoutField('link', 'i'),
)
[docs]
def is_null(self) -> bool:
"""
"""
return PfPageId(self.get('link')) == IxPointer.NULL
[docs]
def set_null(self) -> None:
"""
"""
self.set('link', int(IxPointer.NULL))
[docs]
class IxLeafPointer(IxPointer):
"""
"""
def __repr__(self) -> str: # pragma: no cover
return f"IxLeafPointer({dump(
rid=self.rid,
bid=self.bid,
)})"
[docs]
@classmethod
def new(cls, rid: RmRecordId) -> Self:
"""
"""
data = IxNodeData.new(cls.size)
pointer = cls(data)
pointer.set('rid_pid', int(rid.pid))
pointer.set('rid_sid', int(rid.sid))
pointer.set_null()
return pointer
@property
def rid(self) -> RmRecordId:
"""
"""
return RmRecordId(
pid=RmPageId(self.get('rid_pid')),
sid=RmSlotId(self.get('rid_sid')),
)
@rid.setter
def rid(self, rid: RmRecordId) -> None:
self.set('rid_pid', int(rid.pid))
self.set('rid_sid', int(rid.sid))
@property
def bid(self) -> IxBucketId | None:
"""
"""
if self.is_null():
return None
return IxBucketId(self.get('link'))
@bid.setter
def bid(self, bid: IxBucketId | None) -> None:
if bid is None:
self.set_null()
return
self.set('link', int(bid))
[docs]
class IxBranchPointer(IxPointer):
"""
"""
def __repr__(self) -> str: # pragma: no cover
return f"IxBranchPointer({dump(child_nid=self.child_nid)})"
[docs]
@classmethod
def new(cls, child_nid: 'IxNodeId') -> Self:
"""
"""
data = IxNodeData.new(cls.size)
pointer = cls(data)
pointer.child_nid = child_nid
return pointer
@property
def child_nid(self) -> 'IxNodeId':
"""
"""
return IxNodeId(self.get('link'))
@child_nid.setter
def child_nid(self, nid: 'IxNodeId') -> None:
self.set('link', int(nid))
[docs]
class IxNode(ABC):
"""
Attributes:
nid:
page:
domain:
slot_capacity:
"""
registry: ClassVar[dict[str, type[Self]]] = {}
nid: IxNodeId
page: PfPage
domain: Domain
slot_capacity: int
def __init__(self,
page: PfPage,
domain: Domain,
slot_capacity: int,
) -> None:
self.nid = IxNodeId(page.pid)
self.page = page
self.domain = domain
self.slot_capacity = slot_capacity
@classmethod
def register(cls, kind: str, NodeClass: type[Self]) -> None:
cls.registry[kind] = NodeClass
[docs]
@classmethod
def parse(cls, page: PfPage, domain: Domain, slot_capacity: int) -> Self:
"""
"""
header = IxNodeHeader(
page,
offset=PfPageHeader.size,
limit=IxNodeHeader.size,
)
leaf = header.get('leaf')
kind = 'leaf' if leaf else 'branch'
assert kind in cls.registry
NodeClass = cls.registry[kind]
return NodeClass(page, domain, slot_capacity)
@property
def header_offset(self) -> int:
"""
"""
return self.page.payload_offset
@property
def header_size(self) -> int:
"""
"""
return IxNodeHeader.size
@property
def key_area_offset(self) -> int:
"""
"""
return self.header_offset + self.header_size
@property
def key_size(self) -> int:
"""
"""
return self.domain.size
@property
def key_capacity(self) -> int:
"""
"""
return self.slot_capacity
@property
def key_count(self) -> int:
"""
"""
header = self.get_header()
return header.get('key_count')
@key_count.setter
def key_count(self, count: int) -> None:
header = self.get_header()
header.set('key_count', count)
header.commit()
@property
def pointer_area_offset(self) -> int:
"""
"""
return self.key_area_offset + self.key_capacity * self.key_size
@property
def pointer_size(self) -> int:
"""
"""
return IxPointer.size
@property
def pointer_capacity(self) -> int:
"""
"""
return self.key_capacity + self.slot_count_delta
@property
def pointer_count(self) -> int:
"""
"""
return self.key_count + self.slot_count_delta
@property
@abstractmethod
def PointerClass(self) -> type[IxPointer]: # pragma: no cover
"""
"""
pass
@property
@abstractmethod
def slot_count_delta(self) -> int: # pragma: no cover
"""
"""
pass
@property
def underflow_threshold(self) -> int:
"""
"""
return self.key_capacity // 2
@property
def last_sid(self) -> IxSlotId:
"""
"""
return IxSlotId(self.key_count - 1)
@property
def leaf(self) -> bool:
"""
"""
header = self.get_header()
return header.get('leaf')
@property
def parent_nid(self) -> IxNodeId | None:
"""
"""
header = self.get_header()
nid_ = header.get('parent_nid')
if nid_ == int(IxNodeHeader.NO_PARENT_NODE):
return None
return IxNodeId(nid_)
@parent_nid.setter
def parent_nid(self, nid: IxNodeId | None) -> None:
if nid is None:
nid = IxNodeHeader.NO_PARENT_NODE
header = self.get_header()
header.set('parent_nid', int(nid))
header.commit()
@property
def prev_nid(self) -> IxNodeId | None:
"""
"""
header = self.get_header()
nid_ = header.get('prev_nid')
if nid_ == int(IxNodeHeader.NO_PREV_NODE):
return None
return IxNodeId(nid_)
@prev_nid.setter
def prev_nid(self, nid: IxNodeId | None) -> None:
if nid is None:
nid = IxNodeHeader.NO_PREV_NODE
header = self.get_header()
header.set('prev_nid', int(nid))
header.commit()
@property
def next_nid(self) -> IxNodeId | None:
"""
"""
header = self.get_header()
nid_ = header.get('next_nid')
if nid_ == int(IxNodeHeader.NO_NEXT_NODE):
return None
return IxNodeId(nid_)
@next_nid.setter
def next_nid(self, nid: IxNodeId | None) -> None:
if nid is None: # pragma: no cover
nid = IxNodeHeader.NO_NEXT_NODE
header = self.get_header()
header.set('next_nid', int(nid))
header.commit()
[docs]
def get_key_offset(self, sid: IxSlotId) -> int:
"""
"""
assert 0 <= int(sid) < self.key_capacity
return self.key_area_offset + int(sid) * self.key_size
[docs]
def get_key(self, sid: IxSlotId) -> Value:
"""
"""
assert 0 <= int(sid) < self.key_count
offset = self.get_key_offset(sid)
data_ = self.page.get_data(offset, self.key_size)
return self.domain.decode(data_)
[docs]
def set_key(self, sid: IxSlotId, key: Value) -> None:
"""
"""
assert 0 <= int(sid) < self.key_capacity
offset = self.get_key_offset(sid)
data_ = self.domain.encode(key)
self.page.set_data(offset, PfPageData(data_))
[docs]
def keys(self,
begin: IxSlotId | None = None,
end: IxSlotId | None = None,
) -> Iterator[tuple[IxSlotId, Value]]:
"""
"""
begin_ = int(begin) if begin is not None else 0
end_ = int(end) if end is not None else self.key_count
assert 0 <= begin_ <= end_ <= self.key_count
sids_ = range(begin_, end_)
for sid_ in sids_:
sid = IxSlotId(sid_)
yield sid, self.get_key(sid)
[docs]
def get_pointer_offset(self, sid: IxSlotId) -> int:
"""
"""
assert 0 <= int(sid) < self.pointer_capacity
return self.pointer_area_offset + int(sid) * self.pointer_size
[docs]
def get_pointer(self, sid: IxSlotId) -> IxPointer:
"""
"""
assert 0 <= int(sid) < self.pointer_count
offset = self.get_pointer_offset(sid)
data_ = self.page.get_data(offset, IxPointer.size)
return self.PointerClass(IxNodeData(data_))
[docs]
def set_pointer(self, sid: IxSlotId, pointer: IxPointer) -> None:
"""
"""
assert 0 <= int(sid) < self.pointer_capacity
offset = self.get_pointer_offset(sid)
self.page.set_data(offset, PfPageData(pointer.buffer))
[docs]
def pointers(self,
begin: IxSlotId | None = None,
end: IxSlotId | None = None,
) -> Iterator[tuple[IxSlotId, IxPointer]]:
"""
"""
begin_ = int(begin) if begin is not None else 0
end_ = int(end) if end is not None else self.pointer_count
assert 0 <= begin_ <= end_ <= self.pointer_count
sids_ = range(begin_, end_)
for sid_ in sids_:
sid = IxSlotId(sid_)
yield sid, self.get_pointer(sid)
[docs]
def is_empty(self) -> bool:
"""
"""
return self.key_count == 0
[docs]
def is_full(self) -> bool:
"""
"""
return self.key_count == self.key_capacity
[docs]
def is_underflow(self) -> bool:
"""
"""
return self.key_count < self.underflow_threshold
[docs]
def can_lend(self) -> bool:
"""
"""
return self.key_count > self.underflow_threshold
[docs]
def snapshot(self) -> tuple[list[Value], list[IxPointer]]:
"""
"""
keys = [key for _, key in self.keys()]
pointers = [pointer for _, pointer in self.pointers()]
return keys, pointers
[docs]
def restore(self,
keys: list[Value],
pointers: list[IxPointer],
) -> None:
"""
"""
assert len(keys) + self.slot_count_delta == len(pointers)
for sid_, key in enumerate(keys):
self.set_key(IxSlotId(sid_), key)
for sid_, pointer in enumerate(pointers):
self.set_pointer(IxSlotId(sid_), pointer)
self.key_count = len(keys)
[docs]
@assignment
def search_key(self, value: Value) -> IxSlotId | None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def find_slot(self, value: Value) -> IxSlotId:
"""
"""
raise NotImplementedError
[docs]
@assignment
def unpin(self) -> None:
"""
"""
raise NotImplementedError
[docs]
class IxLeafNode(IxNode):
"""
"""
def __repr__(self) -> str: # pragma: no cover
return f"IxLeafNode({dump(
nid=self.nid,
key_count=self.key_count,
)})"
[docs]
@classmethod
def new(cls, page: PfPage, domain: Domain, slot_capacity: int) -> Self:
"""
"""
node = cls(page, domain, slot_capacity)
header = node.get_header()
header.set('leaf', True)
header.set('key_count', 0)
header.set('parent_nid', int(IxNodeHeader.NO_PARENT_NODE))
header.set('prev_nid', int(IxNodeHeader.NO_PREV_NODE))
header.set('next_nid', int(IxNodeHeader.NO_NEXT_NODE))
header.commit()
return node
@property
def PointerClass(self) -> type[IxPointer]:
return IxLeafPointer
@property
def slot_count_delta(self) -> int:
return 0
[docs]
class IxBranchNode(IxNode):
"""
"""
def __repr__(self) -> str: # pragma: no cover
return f"IxBranchNode({dump(
nid=self.nid,
key_count=self.key_count,
)})"
[docs]
@classmethod
def new(cls,
page: PfPage,
domain: Domain,
slot_capacity: int,
*,
child_nid: IxNodeId,
) -> Self:
"""
"""
node = cls(page, domain, slot_capacity)
header = node.get_header()
header.set('leaf', False)
header.set('key_count', 0)
header.set('parent_nid', int(IxNodeHeader.NO_PARENT_NODE))
header.set('prev_nid', int(IxNodeHeader.NO_PREV_NODE))
header.set('next_nid', int(IxNodeHeader.NO_NEXT_NODE))
header.commit()
pointer = IxBranchPointer.new(child_nid)
node.set_pointer(IxSlotId.first(), pointer)
return node
@property
def PointerClass(self) -> type[IxPointer]:
return IxBranchPointer
@property
def slot_count_delta(self) -> int:
return 1
[docs]
@assignment
def find_pointer(self, nid: IxNodeId) -> IxSlotId:
"""
"""
raise NotImplementedError
IxNode.register('leaf', IxLeafNode)
IxNode.register('branch', IxBranchNode)