memory (sqla): Improve handling of RSM in get_items:
- update queries to handle correctly the `before=""` case and RSM order (older to newer) while having correct order when `max_items` without RSM is used (then order it newer to older). - put RSM code path in a separated method for better readability. - check values of `max_items` and `from_index` and raise exception is they are not positive - when using `before` or `after`, check that the item ID used as reference exists, and raise an exception otherwise. - updated docstring to reflect the new behaviour. - update pubsub cache use of `get_items` with RSM, to adapt to new behaviour. rel 476
This commit is contained in:
Goffi
parent
9cf71f2119
commit
57d3d8e07e
3 changed files with 176 additions and 124 deletions
|
|
@ -78,6 +78,10 @@ class DataError(Exception):
|
|||
pass
|
||||
|
||||
|
||||
class BadRequest(DataError):
|
||||
"""Request contains invalid parameters"""
|
||||
|
||||
|
||||
class ExternalRequestError(Exception):
|
||||
"""Request to third party server failed"""
|
||||
|
||||
|
|
|
|||
|
|
@ -1436,15 +1436,11 @@ class Storage:
|
|||
)
|
||||
|
||||
@aio
|
||||
async def update_pubsub_node(
|
||||
self, node: PubsubNode, values: dict
|
||||
) -> None:
|
||||
async def update_pubsub_node(self, node: PubsubNode, values: dict) -> None:
|
||||
async with self.session() as session:
|
||||
async with session.begin():
|
||||
await session.execute(
|
||||
update(PubsubNode)
|
||||
.filter_by(id=node.id)
|
||||
.values(**values)
|
||||
update(PubsubNode).filter_by(id=node.id).values(**values)
|
||||
)
|
||||
|
||||
@aio
|
||||
|
|
@ -1589,42 +1585,170 @@ class Storage:
|
|||
await session.execute(stmt)
|
||||
await session.commit()
|
||||
|
||||
async def _get_items_rsm(
|
||||
self,
|
||||
node: PubsubNode,
|
||||
max_items: int,
|
||||
before: str | None,
|
||||
after: str | None,
|
||||
from_index: int | None,
|
||||
order: list,
|
||||
metadata: dict,
|
||||
) -> tuple[list[PubsubItem], dict]:
|
||||
"""Get Pubsub Items using RSM
|
||||
|
||||
@param node: retrieve items from this node (must be synchronised)
|
||||
@param max_items: maximum number of items to retrieve
|
||||
@param before: get items which are before the item with this name
|
||||
@param after: get items which are after the item with this name
|
||||
@param from_index: get items starting from this index (None if before/after used)
|
||||
@param order: SQL order clauses
|
||||
@param metadata: metadata dict to populate with RSM information
|
||||
@return: tuple of (items list, metadata dict)
|
||||
@raise exceptions.NotFound: if before or after references a non-existent item.
|
||||
"""
|
||||
# CTE to have result row numbers
|
||||
row_num_q = select(
|
||||
PubsubItem.id,
|
||||
PubsubItem.name,
|
||||
# row_number starts from 1, but RSM index must start from 0
|
||||
(func.row_number().over(order_by=order) - 1).label("item_index"),
|
||||
).filter_by(node_id=node.id)
|
||||
|
||||
row_num_cte = row_num_q.cte()
|
||||
|
||||
if max_items != 0:
|
||||
# As we can't simply use PubsubItem.id when we order by modification, we need
|
||||
# to use row number.
|
||||
|
||||
stmt = (
|
||||
select(row_num_cte.c.item_index, PubsubItem)
|
||||
.join(row_num_cte, PubsubItem.id == row_num_cte.c.id)
|
||||
.limit(max_items)
|
||||
)
|
||||
|
||||
if before == "":
|
||||
# We look for the last page, we need to use descending order for the
|
||||
# query, and reverse later.
|
||||
stmt = stmt.order_by(row_num_cte.c.item_index.desc())
|
||||
elif before or after:
|
||||
# We use an item name as limit, we find which row number correspond to it.
|
||||
item_name = before or after
|
||||
row_num_limit_q = (
|
||||
select(row_num_cte.c.item_index).where(
|
||||
row_num_cte.c.name == item_name
|
||||
)
|
||||
).scalar_subquery()
|
||||
|
||||
# Check if the reference item exists by executing the subquery
|
||||
async with self.session() as session:
|
||||
ref_index_result = await session.execute(row_num_limit_q)
|
||||
ref_index = ref_index_result.scalar_one_or_none()
|
||||
|
||||
if ref_index is None:
|
||||
raise exceptions.NotFound(
|
||||
f"Item {item_name!r} not found in result set"
|
||||
)
|
||||
|
||||
if before:
|
||||
# We order by descending for before, we'll reverse after the
|
||||
# retrieved items.
|
||||
stmt = stmt.where(row_num_cte.c.item_index < ref_index).order_by(
|
||||
row_num_cte.c.item_index.desc()
|
||||
)
|
||||
elif after:
|
||||
stmt = stmt.where(row_num_cte.c.item_index > ref_index).order_by(
|
||||
row_num_cte.c.item_index.asc()
|
||||
)
|
||||
else:
|
||||
# If we reach here, from_index must be set.
|
||||
assert from_index is not None
|
||||
stmt = stmt.where(row_num_cte.c.item_index >= from_index).order_by(
|
||||
row_num_cte.c.item_index.asc()
|
||||
)
|
||||
|
||||
async with self.session() as session:
|
||||
if max_items == 0:
|
||||
items = result = []
|
||||
else:
|
||||
result = await session.execute(stmt)
|
||||
result = result.all()
|
||||
if before is not None:
|
||||
result.reverse()
|
||||
items = [row[-1] for row in result]
|
||||
rows_count = (
|
||||
await session.execute(row_num_q.with_only_columns(count()))
|
||||
).scalar_one()
|
||||
|
||||
try:
|
||||
index = result[0][0]
|
||||
except IndexError:
|
||||
index = None
|
||||
|
||||
try:
|
||||
first = result[0][1].name
|
||||
except IndexError:
|
||||
first = None
|
||||
last = None
|
||||
else:
|
||||
last = result[-1][1].name
|
||||
|
||||
metadata["rsm"] = {
|
||||
k: v
|
||||
for k, v in {
|
||||
"index": index,
|
||||
"count": rows_count,
|
||||
"first": first,
|
||||
"last": last,
|
||||
}.items()
|
||||
if v is not None
|
||||
}
|
||||
metadata["complete"] = (index or 0) + len(result) == rows_count
|
||||
|
||||
return items, metadata
|
||||
|
||||
@aio
|
||||
async def get_items(
|
||||
self,
|
||||
node: PubsubNode,
|
||||
max_items: Optional[int] = None,
|
||||
item_ids: Optional[list[str]] = None,
|
||||
before: Optional[str] = None,
|
||||
after: Optional[str] = None,
|
||||
from_index: Optional[int] = None,
|
||||
order_by: Optional[List[str]] = None,
|
||||
max_items: int | None = None,
|
||||
item_ids: list[str] | None = None,
|
||||
before: str | None = None,
|
||||
after: str | None = None,
|
||||
from_index: int | None = None,
|
||||
order_by: list[str] | None = None,
|
||||
desc: bool = True,
|
||||
force_rsm: bool = False,
|
||||
) -> Tuple[List[PubsubItem], dict]:
|
||||
) -> tuple[list[PubsubItem], dict]:
|
||||
"""Get Pubsub Items from cache
|
||||
|
||||
@param node: retrieve items from this node (must be synchronised)
|
||||
@param max_items: maximum number of items to retrieve
|
||||
@param before: get items which are before the item with this name in given order
|
||||
empty string is not managed here, use desc order to reproduce RSM
|
||||
behaviour.
|
||||
@param item_ids: list of specific item IDs to retrieve.
|
||||
Cannot be used with RSM parameters (before, after, from_index).
|
||||
@param before: get items which are before the item with this name in given order.
|
||||
Use empty string (before="") to request the last page.
|
||||
@param after: get items which are after the item with this name in given order
|
||||
@param from_index: get items with item index (as defined in RSM spec)
|
||||
starting from this number
|
||||
@param order_by: sorting order of items (one of C.ORDER_BY_*)
|
||||
@param desc: direction or ordering
|
||||
@param desc: direction of ordering.
|
||||
If True, ordering is descending.
|
||||
@param force_rsm: if True, force the use of RSM worklow.
|
||||
RSM workflow is automatically used if any of before, after or
|
||||
from_index is used, but if only RSM max_items is used, it won't be
|
||||
used by default. This parameter let's use RSM workflow in this
|
||||
case. Note that in addition to RSM metadata, the result will not be
|
||||
the same (max_items without RSM will returns most recent items,
|
||||
i.e. last items in modification order, while max_items with RSM
|
||||
will return the oldest ones (i.e. first items in modification
|
||||
order).
|
||||
to be used when max_items is used from RSM
|
||||
RSM workflow is automatically used if any of before, after or from_index is
|
||||
used, but if only RSM max_items is used, it won't be used by default. This
|
||||
parameter let use RSM workflow in this case. Note that in addition to RSM
|
||||
metadata, the result will not be the same (max_items without RSM will returns
|
||||
most recent items, i.e. last items in modification order, while max_items with
|
||||
RSM will return the oldest ones (i.e. first items in modification order).
|
||||
To be used when max_items is used from RSM
|
||||
@raise exceptions.BadRequest: Something is wrong with the request.
|
||||
@raise exceptions.NotFound: if before or after references a non-existent item.
|
||||
"""
|
||||
if max_items is not None and max_items < 0:
|
||||
raise exceptions.BadRequest("max_items must be non-negative")
|
||||
if from_index is not None and from_index < 0:
|
||||
raise exceptions.BadRequest("from_index must be non-negative")
|
||||
|
||||
metadata = {
|
||||
"service": node.service,
|
||||
|
|
@ -1638,17 +1762,7 @@ class Storage:
|
|||
if max_items is None:
|
||||
max_items = 20
|
||||
|
||||
use_rsm = any((before, after, from_index is not None))
|
||||
if force_rsm and not use_rsm:
|
||||
#
|
||||
use_rsm = True
|
||||
from_index = 0
|
||||
|
||||
stmt = select(PubsubItem).filter_by(node_id=node.id).limit(max_items)
|
||||
|
||||
if item_ids is not None:
|
||||
stmt = stmt.where(PubsubItem.name.in_(item_ids))
|
||||
|
||||
# ordering
|
||||
if not order_by:
|
||||
order_by = [C.ORDER_BY_MODIFICATION]
|
||||
|
||||
|
|
@ -1667,94 +1781,33 @@ class Storage:
|
|||
else:
|
||||
raise exceptions.InternalError(f"Unknown order type {order_type!r}")
|
||||
|
||||
stmt = stmt.order_by(*order)
|
||||
use_rsm = any((before is not None, after, from_index is not None))
|
||||
if force_rsm and not use_rsm:
|
||||
use_rsm = True
|
||||
from_index = 0
|
||||
|
||||
# item_ids with RSM doesn't make sense
|
||||
if item_ids is not None and use_rsm:
|
||||
raise exceptions.BadRequest(
|
||||
"item_ids cannot be used with RSM parameters"
|
||||
)
|
||||
|
||||
if use_rsm:
|
||||
# CTE to have result row numbers
|
||||
row_num_q = select(
|
||||
PubsubItem.id,
|
||||
PubsubItem.name,
|
||||
# row_number starts from 1, but RSM index must start from 0
|
||||
(func.row_number().over(order_by=order) - 1).label("item_index"),
|
||||
).filter_by(node_id=node.id)
|
||||
return await self._get_items_rsm(
|
||||
node, max_items, before, after, from_index, order, metadata
|
||||
)
|
||||
|
||||
row_num_cte = row_num_q.cte()
|
||||
# Query without RSM
|
||||
stmt = select(PubsubItem).filter_by(node_id=node.id).limit(max_items)
|
||||
stmt = stmt.order_by(*order)
|
||||
|
||||
if max_items > 0:
|
||||
# as we can't simply use PubsubItem.id when we order by modification,
|
||||
# we need to use row number
|
||||
item_name = before or after
|
||||
row_num_limit_q = (
|
||||
select(row_num_cte.c.item_index).where(
|
||||
row_num_cte.c.name == item_name
|
||||
)
|
||||
).scalar_subquery()
|
||||
|
||||
stmt = (
|
||||
select(row_num_cte.c.item_index, PubsubItem)
|
||||
.join(row_num_cte, PubsubItem.id == row_num_cte.c.id)
|
||||
.limit(max_items)
|
||||
)
|
||||
if before:
|
||||
stmt = stmt.where(
|
||||
row_num_cte.c.item_index < row_num_limit_q
|
||||
).order_by(row_num_cte.c.item_index.desc())
|
||||
elif after:
|
||||
stmt = stmt.where(
|
||||
row_num_cte.c.item_index > row_num_limit_q
|
||||
).order_by(row_num_cte.c.item_index.asc())
|
||||
else:
|
||||
stmt = stmt.where(row_num_cte.c.item_index >= from_index).order_by(
|
||||
row_num_cte.c.item_index.asc()
|
||||
)
|
||||
# from_index is used
|
||||
|
||||
async with self.session() as session:
|
||||
if max_items == 0:
|
||||
items = result = []
|
||||
else:
|
||||
result = await session.execute(stmt)
|
||||
result = result.all()
|
||||
if before:
|
||||
result.reverse()
|
||||
items = [row[-1] for row in result]
|
||||
rows_count = (
|
||||
await session.execute(row_num_q.with_only_columns(count()))
|
||||
).scalar_one()
|
||||
|
||||
try:
|
||||
index = result[0][0]
|
||||
except IndexError:
|
||||
index = None
|
||||
|
||||
try:
|
||||
first = result[0][1].name
|
||||
except IndexError:
|
||||
first = None
|
||||
last = None
|
||||
else:
|
||||
last = result[-1][1].name
|
||||
|
||||
metadata["rsm"] = {
|
||||
k: v
|
||||
for k, v in {
|
||||
"index": index,
|
||||
"count": rows_count,
|
||||
"first": first,
|
||||
"last": last,
|
||||
}.items()
|
||||
if v is not None
|
||||
}
|
||||
metadata["complete"] = (index or 0) + len(result) == rows_count
|
||||
|
||||
return items, metadata
|
||||
if item_ids is not None:
|
||||
stmt = stmt.where(PubsubItem.name.in_(item_ids))
|
||||
|
||||
async with self.session() as session:
|
||||
result = await session.execute(stmt)
|
||||
|
||||
result = result.scalars().all()
|
||||
if desc:
|
||||
result.reverse()
|
||||
return result, metadata
|
||||
|
||||
def _get_sqlite_path(self, path: List[Union[str, int]]) -> str:
|
||||
|
|
|
|||
|
|
@ -505,20 +505,15 @@ class PubsubCache:
|
|||
node, max_items=max_items, order_by=extra.get(C.KEY_ORDER_BY)
|
||||
)
|
||||
else:
|
||||
desc = False
|
||||
if rsm_request.before == "":
|
||||
before = None
|
||||
desc = True
|
||||
else:
|
||||
before = rsm_request.before
|
||||
# RSM request
|
||||
pubsub_items, metadata = await self.host.memory.storage.get_items(
|
||||
node,
|
||||
max_items=rsm_request.max,
|
||||
before=before,
|
||||
before=rsm_request.before,
|
||||
after=rsm_request.after,
|
||||
from_index=rsm_request.index,
|
||||
order_by=extra.get(C.KEY_ORDER_BY),
|
||||
desc=desc,
|
||||
desc=False,
|
||||
force_rsm=True,
|
||||
)
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue