from typing import (
Self,
ClassVar,
)
from collections.abc import Iterator
from bluebase.core import (
LayoutField,
LayoutMixin,
dump,
assignment,
)
from bluebase.pf import (
PfPageData,
PfPageView,
PfPage,
)
from bluebase.rm import (
RmRecordId,
RmSlotId,
RmPageId,
)
from .type import (
IxPostingId,
IxBucketId,
IxBucketData,
)
__all__ = (
'IxPosting',
'IxBucketHeader',
'IxBucket',
)
[docs]
class IxPosting(LayoutMixin):
"""
.. rubric:: Layout
.. list-table::
:header-rows: 1
* - Offset
- Size
- Name
- Format
* - 0
- 4
- `rid_pid`
- ``'<i'``
* - 4
- 4
- `rid_sid`
- ``'<i'``
"""
fields = (
LayoutField('rid_pid', 'i'),
LayoutField('rid_sid', 'i'),
)
def __repr__(self) -> str: # pragma: no cover
return f"IxPosting({dump(rid=self.rid)})"
[docs]
@classmethod
def new(cls, rid: RmRecordId) -> Self:
"""
"""
data = IxBucketData.new(cls.size)
posting = cls(data)
posting.set('rid_pid', int(rid.pid))
posting.set('rid_sid', int(rid.sid))
return posting
@property
def rid(self) -> RmRecordId:
"""
"""
return RmRecordId(
RmPageId(self.get('rid_pid')),
RmSlotId(self.get('rid_sid')),
)
[docs]
class IxBucket:
"""
Attributes:
bid:
page:
"""
bid: IxBucketId
page: PfPage
def __repr__(self) -> str: # pragma: no cover
return f"IxBucket({dump(
bid=self.bid,
posting_count=self.posting_count,
)})"
def __init__(self, page: PfPage) -> None:
self.bid = IxBucketId(page.pid)
self.page = page
[docs]
@classmethod
def new(cls, page: PfPage) -> Self:
"""
"""
bucket = cls(page)
header = bucket.get_header()
header.set('posting_count', 0)
header.set('prev_bid', int(IxBucketHeader.NO_PREV_BUCKET))
header.set('next_bid', int(IxBucketHeader.NO_NEXT_BUCKET))
header.commit()
return bucket
@property
def header_offset(self) -> int:
"""
"""
return self.page.payload_offset
@property
def header_size(self) -> int:
"""
"""
return IxBucketHeader.size
@property
def payload_offset(self) -> int:
"""
"""
return self.header_offset + self.header_size
@property
def payload_size(self) -> int:
"""
"""
return self.page.size - self.payload_offset
@property
def posting_size(self) -> int:
"""
"""
return IxPosting.size
@property
def posting_capacity(self) -> int:
"""
"""
return self.payload_size // self.posting_size
@property
def posting_count(self) -> int:
"""
"""
header = self.get_header()
return header.get('posting_count')
@posting_count.setter
def posting_count(self, count: int) -> None:
header = self.get_header()
header.set('posting_count', count)
header.commit()
@property
def prev_bid(self) -> IxBucketId | None:
"""
"""
header = self.get_header()
bid_ = header.get('prev_bid')
if bid_ == int(IxBucketHeader.NO_PREV_BUCKET):
return None
return IxBucketId(bid_)
@prev_bid.setter
def prev_bid(self, bid: IxBucketId | None) -> None:
if bid is None:
bid = IxBucketHeader.NO_PREV_BUCKET
header = self.get_header()
header.set('prev_bid', int(bid))
header.commit()
@property
def next_bid(self) -> IxBucketId | None:
"""
"""
header = self.get_header()
bid_ = header.get('next_bid')
if bid_ == int(IxBucketHeader.NO_NEXT_BUCKET):
return None
return IxBucketId(bid_)
@next_bid.setter
def next_bid(self, bid: IxBucketId | None) -> None:
if bid is None:
bid = IxBucketHeader.NO_NEXT_BUCKET
header = self.get_header()
header.set('next_bid', int(bid))
header.commit()
@property
def last_pid(self) -> IxPostingId:
"""
"""
return IxPostingId(self.posting_count - 1)
[docs]
def get_posting_offset(self, pid: IxPostingId) -> int:
"""
"""
assert 0 <= int(pid) < self.posting_capacity
return self.payload_offset + int(pid) * self.posting_size
[docs]
def get_posting(self, pid: IxPostingId) -> IxPosting:
"""
"""
assert 0 <= int(pid) < self.posting_count
offset = self.get_posting_offset(pid)
data_ = self.page.get_data(offset, self.posting_size)
return IxPosting(IxBucketData(data_))
[docs]
def set_posting(self, pid: IxPostingId, posting: IxPosting) -> None:
"""
"""
assert 0 <= int(pid) < self.posting_capacity
offset = self.get_posting_offset(pid)
self.page.set_data(offset, PfPageData(posting.buffer))
[docs]
def postings(self,
begin: IxPostingId | None = None,
end: IxPostingId | None = None,
) -> Iterator[tuple[IxPostingId, IxPosting]]:
"""
"""
begin_ = int(begin) if begin is not None else 0
end_ = int(end) if end is not None else self.posting_count
assert 0 <= begin_ <= end_ <= self.posting_count
pids_ = range(begin_, end_)
for pid_ in pids_:
pid = IxPostingId(pid_)
yield pid, self.get_posting(pid)
[docs]
def is_empty(self) -> bool:
"""
"""
return self.posting_count == 0
[docs]
def is_full(self) -> bool:
"""
"""
return self.posting_count == self.posting_capacity
[docs]
@assignment
def search(self, rid: RmRecordId) -> IxPostingId | None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def insert(self, rid: RmRecordId) -> None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def delete(self, rid: RmRecordId) -> None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def unpin(self) -> None:
"""
"""
raise NotImplementedError