from dataclasses import dataclass
from collections.abc import Iterator
from bluebase.core import (
Comp,
Predicate,
Value,
dump,
optional,
assignment,
)
from bluebase.rm import RmRecordId
from .error import IxUnsupportedCompError
from .index import IxIndex
from .type import (
IxNodeId,
IxSlotId,
IxBucketId,
IxPostingId,
IxEntry,
)
from .node import (
IxLeafPointer,
IxBranchPointer,
)
__all__ = (
'IxNodeCursor',
'IxBucketCursor',
'IxIndexScan',
)
[docs]
@dataclass(slots=True)
class IxNodeCursor:
"""
Attributes:
nid:
sid:
"""
nid: IxNodeId
sid: IxSlotId
[docs]
@dataclass(slots=True)
class IxBucketCursor:
"""
Attributes:
bid:
pid:
"""
bid: IxBucketId
pid: IxPostingId
[docs]
class IxIndexScan:
"""
Attributes:
index:
predicate:
node_cursor:
bucket_cursor:
last_scanned:
Raises:
IxUnsupportedCompError:
TypeError:
"""
index: IxIndex
predicate: Predicate | None
node_cursor: IxNodeCursor | None
bucket_cursor: IxBucketCursor | None
last_scanned: IxEntry | None
def __repr__(self) -> str: # pragma: no cover
return f"IxIndexScan({dump(
index=self.index,
predicate=self.predicate,
)})"
def __init__(self,
index: IxIndex,
predicate: Predicate | None = None,
) -> None:
if predicate is not None:
if predicate.comp == Comp.NE:
raise IxUnsupportedCompError(comp=predicate.comp)
index.domain.validate(predicate.value)
self.index = index
self.predicate = predicate
self.node_cursor = None
self.bucket_cursor = None
self.last_scanned = None
[docs]
def __iter__(self) -> Iterator[RmRecordId]:
"""
index scan을 순회하는 iterator를 반환합니다.
"""
return self
[docs]
@assignment
def __next__(self) -> RmRecordId:
"""
Raises:
StopIteration:
"""
raise NotImplementedError
[docs]
@optional
def _init_cursor(self) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _locate_leaf(self, nid: IxNodeId) -> IxNodeId:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _find_slot(self, nid: IxNodeId) -> IxSlotId:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _advance_cursor(self) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _advance_node_cursor(self) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _advance_bucket_cursor(self) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _fetch_entry(self) -> IxEntry:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _fetch_entry_from_node(self) -> IxEntry:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _fetch_entry_from_bucket(self) -> IxEntry:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _match_value(self, value: Value) -> bool:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError