import struct
from abc import (
ABC,
abstractmethod,
)
from collections.abc import Iterator
from bluebase.core import (
Value,
Predicate,
dump,
optional,
assignment,
)
from bluebase.rm import (
RmRecord,
RmFile,
RmPredicate,
RmFileScan,
)
from bluebase.ix import (
IxIndex,
IxIndexScan,
)
from .error import (
QlOpenedOpOpenError,
QlClosedOpAccessError,
)
from .type import (
QlSchema,
QlContext,
QlAttr,
QlRow,
QlCond,
QlAttrCond,
QlValueCond,
)
__all__ = (
'QlOp',
'QlLeafOp',
'QlBranchOp',
'QlFileScanOp',
'QlIndexScanOp',
'QlCartesianOp',
'QlLoopJoinOp',
'QlHashJoinOp',
'QlIndexJoinOp',
'QlFilterOp',
'QlProjectOp',
)
[docs]
class QlOp(ABC):
"""
Attributes:
refs:
opened:
executed:
loop_count:
row_count:
"""
refs: set[str]
opened: bool
executed: bool
loop_count: int
row_count: int
def __init__(self, refs: set[str]) -> None:
self.refs = refs
self.opened = False
self.executed = False
self.loop_count = 0
self.row_count = 0
[docs]
def __iter__(self) -> Iterator[QlRow]:
"""
op를 순회하는 iterator를 반환합니다.
"""
self.open()
try:
while True:
row = self.next()
if row is None:
break
yield row
finally:
self.close()
[docs]
@staticmethod
def get_row(ref: str, schema: QlSchema, record: RmRecord) -> QlRow:
"""
"""
lookup = {}
for attr_info in schema.attrs.values():
data = record.data[attr_info.slice()]
value = attr_info.domain.decode(data)
attr = QlAttr(
ref=ref,
name=attr_info.name,
domain=attr_info.domain,
)
lookup[attr] = value
return QlRow(record, lookup)
@property
@abstractmethod
def attrs(self) -> list[QlAttr]: # pragma: no cover
"""
"""
pass
[docs]
def open(self) -> None:
"""
Raises:
QlOpenedOpOpenError:
"""
if self.opened:
raise QlOpenedOpOpenError(op=self)
self.opened = True
self.loop_count += 1
[docs]
def close(self) -> None:
"""
Raises:
QlClosedOpAccessError:
"""
if not self.opened:
raise QlClosedOpAccessError(op=self)
self.opened = False
[docs]
def next(self) -> QlRow | None:
"""
Raises:
QlClosedOpAccessError
StopIteration
"""
if not self.opened:
raise QlClosedOpAccessError(op=self)
self.executed = True
row = self.get_next()
if row is not None:
self.row_count += 1
return row
[docs]
@abstractmethod
def get_next(self) -> QlRow | None: # pragma: no cover
"""
"""
pass
[docs]
def analysis(self) -> str: # pragma: no cover
"""
"""
if self.executed:
return f" ({self.loop_count} loops, {self.row_count} rows)"
return ""
[docs]
@abstractmethod
def explain(self, depth: int = 0) -> str: # pragma: no cover
"""
"""
pass
[docs]
class QlLeafOp(QlOp, ABC):
"""
Attributes:
context:
ref:
cond:
"""
context: QlContext
ref: str
cond: QlValueCond | None
def __init__(self,
context: QlContext,
ref: str,
cond: QlValueCond | None = None,
) -> None:
super().__init__({ref})
self.context = context
self.ref = ref
self.cond = cond
@property
def schema(self) -> QlSchema:
"""
"""
return self.context.schemas[self.ref]
[docs]
class QlBranchOp(QlOp, ABC):
"""
"""
pass
[docs]
class QlFileScanOp(QlLeafOp):
"""
Attributes:
file:
scan:
predicate:
"""
file: RmFile | None
scan: RmFileScan | None
predicate: RmPredicate | None
def __repr__(self) -> str: # pragma: no cover
return f"QlFileScanOp({dump(ref=self.ref, cond=self.cond)})"
def __init__(self,
context: QlContext,
ref: str,
cond: QlValueCond | None = None,
) -> None:
super().__init__(context, ref, cond)
if cond is None:
self.predicate = None
else:
attr_info = self.schema.attrs[cond.left_attr.name]
self.predicate = RmPredicate(
name=attr_info.name,
comp=cond.comp,
value=cond.right_value,
domain=attr_info.domain,
offset=attr_info.offset,
)
self.file = None
self.scan = None
@property
def attrs(self) -> list[QlAttr]:
return [QlAttr(
ref=self.ref,
name=attr_info.name,
domain=attr_info.domain,
) for attr_info in self.schema.attrs.values()]
[docs]
@assignment
def open(self) -> None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def close(self) -> None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def get_next(self) -> QlRow | None:
"""
"""
raise NotImplementedError
[docs]
def explain(self, depth: int = 0) -> str: # pragma: no cover
return "".join([
" " * depth,
"FileScan(",
f"{self.schema.table.name}",
"" if self.predicate is None else f", {self.predicate.verbose()}",
f"):{self.ref}",
self.analysis(),
"\n",
])
[docs]
class QlIndexScanOp(QlLeafOp):
"""
Attributes:
file:
index:
scan:
predicate:
"""
file: RmFile | None
index: IxIndex | None
scan: IxIndexScan | None
predicate: Predicate
def __repr__(self) -> str: # pragma: no cover
return f"QlIndexScanOp({dump(ref=self.ref, cond=self.cond)})"
def __init__(self, context: QlContext, ref: str, cond: QlValueCond) -> None:
super().__init__(context, ref, cond)
self.predicate = Predicate(
name=self.schema.attrs[cond.left_attr.name].name,
comp=cond.comp,
value=cond.right_value,
)
self.file = None
self.index = None
self.scan = None
@property
def attrs(self) -> list[QlAttr]:
return [QlAttr(
ref=self.ref,
name=attr_info.name,
domain=attr_info.domain,
) for attr_info in self.schema.attrs.values()]
[docs]
@assignment
def open(self) -> None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def close(self) -> None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def get_next(self) -> QlRow | None:
"""
"""
raise NotImplementedError
[docs]
def explain(self, depth: int = 0) -> str: # pragma: no cover
return "".join([
" " * depth,
"IndexScan(",
f"{self.schema.table.name}",
f", {self.predicate.verbose()}",
f"):{self.ref}",
self.analysis(),
"\n",
])
[docs]
class QlCartesianOp(QlBranchOp):
"""
Attributes:
left_op:
right_op:
left_row:
"""
left_op: QlOp
right_op: QlLeafOp
left_row: QlRow | None
def __repr__(self) -> str: # pragma: no cover
return f"QlCartesianOp({dump(
left_op=self.left_op,
right_op=self.right_op,
)})"
def __init__(self, left_op: QlOp, right_op: QlLeafOp) -> None:
super().__init__(left_op.refs | right_op.refs)
self.left_op = left_op
self.right_op = right_op
self.left_row = None
@property
def attrs(self) -> list[QlAttr]:
return self.left_op.attrs + self.right_op.attrs
[docs]
@optional
def _advance(self) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@assignment
def open(self) -> None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def close(self) -> None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def get_next(self) -> QlRow | None:
"""
"""
raise NotImplementedError
[docs]
def explain(self, depth: int = 0) -> str: # pragma: no cover
return "".join([
" " * depth,
"Cartesian()",
self.analysis(),
"\n",
self.left_op.explain(depth + 1),
self.right_op.explain(depth + 1),
])
[docs]
class QlLoopJoinOp(QlBranchOp):
"""
Attributes:
left_op:
right_op:
on:
left_row:
"""
left_op: QlOp
right_op: QlLeafOp
on: QlAttrCond
left_row: QlRow | None
def __repr__(self) -> str: # pragma: no cover
return f"QlLoopJoinOp({dump(
left_op=self.left_op,
right_op=self.right_op,
on=self.on,
)})"
def __init__(self,
left_op: QlOp,
right_op: QlLeafOp,
on: QlAttrCond,
) -> None:
super().__init__(left_op.refs | right_op.refs)
self.left_op = left_op
self.right_op = right_op
self.on = on
self.left_row = None
@property
def attrs(self) -> list[QlAttr]:
return self.left_op.attrs + self.right_op.attrs
[docs]
@optional
def _advance(self) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@assignment
def open(self) -> None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def close(self) -> None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def get_next(self) -> QlRow | None:
"""
"""
raise NotImplementedError
[docs]
def explain(self, depth: int = 0) -> str: # pragma: no cover
return "".join([
" " * depth,
"LoopJoin(",
self.on.verbose(),
")",
self.analysis(),
"\n",
self.left_op.explain(depth + 1),
self.right_op.explain(depth + 1),
])
[docs]
class QlHashJoinOp(QlBranchOp):
"""
Attributes:
probe_op:
probe_attr:
build_op:
build_attr:
on:
probe_row:
build_rows:
lookup:
"""
probe_op: QlOp
probe_attr: QlAttr
build_op: QlLeafOp
build_attr: QlAttr
on: QlAttrCond
probe_row: QlRow | None
build_rows: list[QlRow]
lookup: dict[bytes, list[QlRow]] | None
def __repr__(self) -> str: # pragma: no cover
return f"QlHashJoinOp({dump(
probe_op=self.probe_op,
probe_attr=self.probe_attr,
build_op=self.build_op,
build_attr=self.build_attr,
on=self.on,
)})"
def __init__(self,
probe_op: QlOp,
probe_attr: QlAttr,
build_op: QlLeafOp,
build_attr: QlAttr,
on: QlAttrCond,
) -> None:
super().__init__(probe_op.refs | build_op.refs)
self.probe_op = probe_op
self.probe_attr = probe_attr
self.build_op = build_op
self.build_attr = build_attr
self.on = on
self.probe_row = None
self.build_rows = []
self.lookup = None
@property
def attrs(self) -> list[QlAttr]:
return self.probe_op.attrs + self.build_op.attrs
[docs]
@optional
def _advance(self) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _pack(self, value: Value) -> bytes:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@optional
def _build(self) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@assignment
def open(self) -> None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def close(self) -> None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def get_next(self) -> QlRow | None:
"""
"""
raise NotImplementedError
[docs]
def explain(self, depth: int = 0) -> str: # pragma: no cover
return "".join([
" " * depth,
"HashJoin(",
self.on.verbose(),
")",
self.analysis(),
"\n",
self.probe_op.explain(depth + 1),
self.build_op.explain(depth + 1),
])
[docs]
class QlIndexJoinOp(QlBranchOp):
"""
Attributes:
outer_op:
outer_attr:
inner_op:
inner_attr:
on:
file:
index:
scan:
outer_row:
"""
outer_op: QlOp
outer_attr: QlAttr
inner_op: QlLeafOp
inner_attr: QlAttr
on: QlCond
file: RmFile | None
index: IxIndex | None
scan: IxIndexScan | None
outer_row: QlRow | None
def __repr__(self) -> str: # pragma: no cover
return f"QlIndexJoinOp({dump(
outer_op=self.outer_op,
outer_attr=self.outer_attr,
inner_op=self.inner_op,
inner_attr=self.inner_attr,
on=self.on,
)})"
def __init__(self,
context: QlContext,
outer_op: QlOp,
outer_attr: QlAttr,
inner_op: QlLeafOp,
inner_attr: QlAttr,
on: QlAttrCond,
) -> None:
super().__init__(outer_op.refs | inner_op.refs)
self.context = context
self.outer_op = outer_op
self.outer_attr = outer_attr
self.inner_op = inner_op
self.inner_attr = inner_attr
self.on = on
self.file = None
self.index = None
self.scan = None
self.outer_row = None
@property
def attrs(self) -> list[QlAttr]:
return self.outer_op.attrs + self.inner_op.attrs
[docs]
@optional
def _advance(self) -> None:
"""
Hint:
이런 함수가 있으면 편하다!
"""
raise NotImplementedError
[docs]
@assignment
def open(self) -> None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def close(self) -> None:
"""
"""
raise NotImplementedError
[docs]
@assignment
def get_next(self) -> QlRow | None:
"""
"""
raise NotImplementedError
[docs]
def explain(self, depth: int = 0) -> str: # pragma: no cover
return "".join([
" " * depth,
"IndexJoin(",
self.on.verbose(),
")",
self.analysis(),
"\n",
self.outer_op.explain(depth + 1),
self.inner_op.explain(depth + 1),
])
[docs]
class QlFilterOp(QlBranchOp):
"""
Attributes:
child:
conds:
"""
child: QlOp
conds: list[QlCond]
def __repr__(self) -> str: # pragma: no cover
return f"QlFilterOp({dump(child=self.child, conds=self.conds)})"
def __init__(self, child: QlOp, conds: list[QlCond]) -> None:
super().__init__(child.refs)
self.child = child
self.conds = conds
@property
def attrs(self) -> list[QlAttr]:
return self.child.attrs
[docs]
def open(self) -> None:
super().open()
self.child.open()
[docs]
def close(self) -> None:
super().close()
self.child.close()
[docs]
@assignment
def get_next(self) -> QlRow | None:
"""
"""
raise NotImplementedError
[docs]
def explain(self, depth: int = 0) -> str: # pragma: no cover
return "".join([
" " * depth,
"Filter(",
", ".join([cond.verbose() for cond in self.conds]),
")",
self.analysis(),
"\n",
self.child.explain(depth + 1),
])
[docs]
class QlProjectOp(QlBranchOp):
"""
Attributes:
child:
"""
child: QlOp
def __repr__(self) -> str: # pragma: no cover
return f"QlProjectOp({dump(child=self.child, attrs=self.attrs)})"
def __init__(self, child: QlOp, attrs: list[QlAttr]) -> None:
super().__init__(child.refs)
self.child = child
self._attrs = attrs
@property
def attrs(self) -> list[QlAttr]:
return self._attrs
[docs]
def open(self) -> None:
super().open()
self.child.open()
[docs]
def close(self) -> None:
super().close()
self.child.close()
[docs]
@assignment
def get_next(self) -> QlRow | None:
"""
"""
raise NotImplementedError
[docs]
def explain(self, depth: int = 0) -> str: # pragma: no cover
return "".join([
" " * depth,
"Project(",
", ".join([f'{attr.ref}.{attr.name}' for attr in self.attrs]),
")",
self.analysis(),
"\n",
self.child.explain(depth + 1),
])