from typing import (
Self,
ClassVar,
)
from bluebase.core import (
LayoutField,
LayoutMixin,
Value,
Domain,
dump,
optional,
assignment,
)
from bluebase.pf import (
PfPageId,
PfPageView,
PfPageHeader,
PfPage,
PfFileHeader,
PfFile,
)
from bluebase.rm import RmRecordId
from .error import IxEntryNotFoundError
from .type import (
IxSlotId,
IxNodeId,
IxBucketId,
IxEntry,
)
from .bucket import IxBucket
from .node import (
IxLeafPointer,
IxBranchPointer,
IxNode,
IxLeafNode,
IxBranchNode,
)
__all__ = (
'IxIndexHeader',
'IxIndex',
)
[docs]
class IxIndex:
"""
Attributes:
file:
domain:
slot_capacity:
"""
file: PfFile
domain: Domain
slot_capacity: int
def __repr__(self) -> str: # pragma: no cover
return f"IxIndex({dump(
file=self.file,
domain=self.domain,
slot_capacity=self.slot_capacity,
)})"
def __init__(self,
file: PfFile,
domain: Domain,
slot_capacity: int,
) -> None:
self.file = file
self.domain = domain
self.slot_capacity = slot_capacity
[docs]
@classmethod
def new(cls, file: PfFile, domain: Domain, slot_capacity: int) -> Self:
"""
"""
page = file.allocate_page()
header = IxIndex.get_header(page)
header.set('domain_enum', domain.enum)
header.set('domain_size', domain.size)
header.set('slot_capacity', slot_capacity)
header.set('root_nid', int(IxIndexHeader.NO_ROOT_NODE))
header.commit()
page.unpin()
return cls(file, domain, slot_capacity)
[docs]
@classmethod
def open(cls, file: PfFile) -> Self:
"""
"""
page, header = IxIndex.fetch_header(file)
domain = Domain.restore(
header.get('domain_enum'),
header.get('domain_size'),
)
slot_capacity = header.get('slot_capacity')
page.unpin()
return cls(file, domain, slot_capacity)
@property
def root_nid(self) -> IxNodeId | None:
"""
"""
page, header = IxIndex.fetch_header(self.file)
nid_ = header.get('root_nid')
page.unpin()
if nid_ == int(IxIndexHeader.NO_ROOT_NODE):
return None
return IxNodeId(nid_)
@root_nid.setter
def root_nid(self, nid: IxNodeId | None) -> None:
if nid is None:
nid = IxIndexHeader.NO_ROOT_NODE
page, header = IxIndex.fetch_header(self.file)
header.set('root_nid', int(nid))
header.commit()
page.unpin()
[docs]
@optional
def _get_node(self, nid: IxNodeId) -> IxNode:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _allocate_leaf(self) -> IxLeafNode:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _allocate_branch(self, *, child_nid: IxNodeId) -> IxBranchNode:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _dispose_node(self, nid: IxNodeId) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _get_bucket(self, bid: IxBucketId) -> IxBucket:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _allocate_bucket(self) -> IxBucket:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _dispose_bucket(self, bid: IxBucketId) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _bucket_insert(self, bid: IxBucketId, rid: RmRecordId) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _bucket_delete(self,
bid: IxBucketId,
rid: RmRecordId,
) -> IxBucketId | None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _locate_leaf(self, nid: IxNodeId, *, value: Value) -> IxNodeId:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _leaf_insert(self,
nid: IxNodeId,
*,
value: Value,
rid: RmRecordId,
) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _branch_insert(self,
nid: IxNodeId,
value: Value,
child_nid: IxNodeId,
) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _push_key_up(self,
value: Value,
*,
node_nid: IxNodeId,
new_node_nid: IxNodeId,
) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _leaf_delete(self,
nid: IxNodeId,
*,
value: Value,
rid: RmRecordId,
) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _fix_key_up(self, nid: IxNodeId, *, min_value: Value) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _find_left_sibling(self, nid: IxNodeId) -> IxNodeId | None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _find_right_sibling(self, nid: IxNodeId) -> IxNodeId | None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _leaf_borrow_left(self,
nid: IxNodeId,
sibling_nid: IxNodeId,
) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _leaf_borrow_right(self,
nid: IxNodeId,
sibling_nid: IxNodeId,
) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _leaf_merge(self, left_nid: IxNodeId, right_nid: IxNodeId) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _branch_delete(self, nid: IxNodeId, *, child_nid: IxNodeId) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _branch_borrow_left(self,
nid: IxNodeId,
sibling_nid: IxNodeId,
) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _branch_borrow_right(self,
nid: IxNodeId,
sibling_nid: IxNodeId,
) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _branch_merge(self, left_nid: IxNodeId, right_nid: IxNodeId) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@assignment
def insert(self, entry: IxEntry) -> None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def delete(self, entry: IxEntry) -> None:
"""
Raises:
IxEntryNotFoundError:
"""
raise NotImplementedError
[docs]
@assignment
def persist(self) -> None:
"""
"""
raise NotImplementedError
if __debug__: # pragma: no cover
def _debug_dump_dot(self, show_data_pid: bool = False) -> str:
if self.root_nid is None:
return 'digraph Ix {}'
F_NAME = 'Menlo'
C_PID = '#0000FF'
C_SID = '#00AA00'
C_LEAF = '#EEFFFF'
C_BRANCH = '#FFFFEE'
C_KEY = '#FF0000'
C_POINTER = '#DDDDDD'
C_BUCKET = '#FFEEFF'
C_DASHED = '#AAAAAA'
S_RID = 12
def P(pid: PfPageId) -> str:
return f'<FONT COLOR="{C_PID}">{pid}</FONT>'
def R(rid: RmRecordId) -> str:
if show_data_pid:
return (
f'<FONT POINT-SIZE="{S_RID}" COLOR="{C_PID}">{rid.pid}</FONT>'
f'<FONT POINT-SIZE="{S_RID}">:</FONT>'
f'<FONT POINT-SIZE="{S_RID}" COLOR="{C_SID}">{rid.sid}</FONT>'
)
else:
return f'<FONT POINT-SIZE="{S_RID}" COLOR="{C_SID}">{rid.sid}</FONT>'
lines = [
'digraph Ix {',
' rankdir=TB;',
f' node [shape=plaintext,fontname="{F_NAME}"];',
f' edge [fontname="{F_NAME}"];',
]
stack = [(self.root_nid, 0)]
depth2nids: dict[int, list[IxNodeId]] = {}
seen = set()
while stack:
nid, depth = stack.pop()
if nid in seen:
continue
seen.add(nid)
if depth not in depth2nids:
depth2nids[depth] = []
depth2nids[depth].append(nid)
node = self._get_node(nid)
keys, pointers = node.snapshot()
cols = []
if node.leaf:
name = 'L'
color = C_LEAF
for sid, key in node.keys():
pointer = node.get_pointer(sid)
assert isinstance(pointer, IxLeafPointer)
rid = pointer.rid
cols.append(f'<TD BGCOLOR="{C_POINTER}" PORT="p{sid}">{R(rid)}</TD>')
cols.append(f'<TD><FONT COLOR="{C_KEY}">{key}</FONT></TD>')
else:
name = 'I'
color = C_BRANCH
for sid, key in node.keys():
cols.append(f'<TD BGCOLOR="{C_POINTER}" PORT="p{sid}"></TD>')
cols.append(f'<TD><FONT COLOR="{C_KEY}">{key}</FONT></TD>')
cols.append(f'<TD BGCOLOR="{C_POINTER}" PORT="p{len(keys)}"></TD>')
table = (
'<TABLE BORDER="1" CELLBORDER="1" CELLSPACING="0">'
f'<TR><TD COLSPAN="{2 * len(keys) + 1}" BGCOLOR="{color}">{name}({P(nid)})</TD></TR>'
f'<TR>{"".join(cols)}</TR>'
'</TABLE>'
)
lines.append(f' n{nid} [label=<{table}>];')
if not node.leaf:
for pid, pointer in enumerate(pointers):
assert isinstance(pointer, IxBranchPointer)
lines.append(f' n{nid}:p{pid} -> n{pointer.child_nid};')
stack.append((pointer.child_nid, depth + 1))
node.unpin()
for _, nids in depth2nids.items():
lines.append(' { rank=same; %s; }' % '; '.join(f'n{nid}' for nid in nids))
leaf_nids = depth2nids[max(depth2nids.keys())]
depth2bids: dict[int, list[IxBucketId]] = {}
for nid in leaf_nids:
node = self._get_node(nid)
assert isinstance(node, IxLeafNode)
if node.next_nid is not None:
lines.append(
f' n{nid}:p{node.last_sid + 1} -> n{node.next_nid} '
f'[style=dashed,color="{C_DASHED}",constraint=false];'
)
for sid, pointer in node.pointers():
assert isinstance(pointer, IxLeafPointer)
if pointer.bid is None:
continue
prev_bid = None
bid = pointer.bid
depth = 0
while True:
if depth not in depth2bids:
depth2bids[depth] = []
depth2bids[depth].append(bid)
bucket = self._get_bucket(bid)
rows = []
for _, posting in bucket.postings():
rid = posting.rid
rows.append(f'<TR><TD>{R(rid)}</TD></TR>')
table = (
'<TABLE BORDER="1" CELLBORDER="1" CELLSPACING="0">'
f'<TR><TD BGCOLOR="{C_BUCKET}">B({P(bid)})</TD></TR>'
f'{"".join(rows)}'
'</TABLE>'
)
lines.append(f' b{bid} [label=<{table}>];')
if prev_bid is None:
lines.append(
f' n{nid}:p{sid} -> b{bid} '
f'[constraint=true];'
)
else:
lines.append(
f' b{prev_bid} -> b{bid} '
f'[constraint=true];'
)
if bucket.next_bid is None:
bucket.unpin()
break
prev_bid = bid
bid = bucket.next_bid
bucket.unpin()
depth += 1
node.unpin()
if depth2bids:
for _, bids in depth2bids.items():
lines.append(' { rank=same; %s; }' % '; '.join(f'b{bid}' for bid in bids))
lines.append('}')
return '\n'.join(lines)
def _debug_validate_invariant(self) -> None:
# root가 없는 경우, 즉시 종료
if self.root_nid is None:
return
# root부터 순회 시작
stack = [self.root_nid]
seen = set()
while stack:
nid = stack.pop()
if nid in seen:
continue
seen.add(nid)
node = self._get_node(nid)
keys, pointers = node.snapshot()
values = [key for key in keys]
# key 순서 검증
for i in range(len(values) - 1):
assert values[i] < values[i + 1] # type: ignore
# leaf의 경우
if node.leaf:
# key, pointer 수 검증
assert len(keys) == len(pointers)
# right 검증
if node.next_nid is not None:
right = self._get_node(node.next_nid)
assert right.prev_nid == node.nid
right_keys, _ = right.snapshot()
right.unpin()
assert keys[-1] < right_keys[0] # type: ignore
# pointer 검증
for pointer in pointers:
assert isinstance(pointer, IxLeafPointer)
# branch의 경우
else:
# key, pointer 수 검증
assert len(keys) + 1 == len(pointers)
# pointer 검증
for key, pointer in zip(keys, pointers):
assert isinstance(pointer, IxBranchPointer)
child = self._get_node(pointer.child_nid)
assert child.parent_nid == node.nid
child_keys, _ = child.snapshot()
child.unpin()
stack.append(child.nid)
# key 순서 검증
assert child_keys[-1] < key # type: ignore
# last pointer 검증
key = keys[-1]
pointer = pointers[-1]
assert isinstance(pointer, IxBranchPointer)
child = self._get_node(pointer.child_nid)
assert child.parent_nid == node.nid
child_keys, _ = child.snapshot()
child.unpin()
stack.append(child.nid)
# key 순서 검증
assert key <= child_keys[0] # type: ignore
node.unpin()
# 가장 왼쪽 node들은 항상 prev_nid가 None
nid = self.root_nid
while True:
node = self._get_node(nid)
assert node.prev_nid is None
if node.leaf:
node.unpin()
break
_, pointers = node.snapshot()
node.unpin()
pointer = pointers[0]
assert isinstance(pointer, IxBranchPointer)
nid = pointer.child_nid
# 가장 오른쪽 node들은 항상 next_nid가 None
nid = self.root_nid
while True:
node = self._get_node(nid)
assert node.next_nid is None
if node.leaf:
node.unpin()
break
_, pointers = node.snapshot()
node.unpin()
pointer = pointers[-1]
assert isinstance(pointer, IxBranchPointer)
nid = pointer.child_nid