aggregates.py 10 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316
  1. """
  2. Aggregation functions
  3. """
  4. from __future__ import annotations
  5. from decimal import Decimal
  6. from typing import (
  7. Any,
  8. Callable,
  9. Dict,
  10. Iterable,
  11. List,
  12. Mapping,
  13. MutableMapping,
  14. Optional,
  15. Set,
  16. Tuple,
  17. TypeVar,
  18. Union,
  19. overload,
  20. )
  21. from rdflib.namespace import XSD
  22. from rdflib.plugins.sparql.datatypes import type_promotion
  23. from rdflib.plugins.sparql.evalutils import _eval, _val
  24. from rdflib.plugins.sparql.operators import numeric
  25. from rdflib.plugins.sparql.parserutils import CompValue
  26. from rdflib.plugins.sparql.sparql import FrozenBindings, NotBoundError, SPARQLTypeError
  27. from rdflib.term import BNode, Identifier, Literal, URIRef, Variable
  28. class Accumulator:
  29. """abstract base class for different aggregation functions"""
  30. def __init__(self, aggregation: CompValue):
  31. self.get_value: Callable[[], Optional[Literal]]
  32. self.update: Callable[[FrozenBindings, Aggregator], None]
  33. self.var = aggregation.res
  34. self.expr = aggregation.vars
  35. if not aggregation.distinct:
  36. # type error: Cannot assign to a method
  37. self.use_row = self.dont_care # type: ignore[method-assign]
  38. self.distinct = False
  39. else:
  40. self.distinct = aggregation.distinct
  41. self.seen: Set[Any] = set()
  42. def dont_care(self, row: FrozenBindings) -> bool:
  43. """skips distinct test"""
  44. return True
  45. def use_row(self, row: FrozenBindings) -> bool:
  46. """tests distinct with set"""
  47. return _eval(self.expr, row) not in self.seen
  48. def set_value(self, bindings: MutableMapping[Variable, Identifier]) -> None:
  49. """sets final value in bindings"""
  50. # type error: Incompatible types in assignment (expression has type "Optional[Literal]", target has type "Identifier")
  51. bindings[self.var] = self.get_value() # type: ignore[assignment]
  52. class Counter(Accumulator):
  53. def __init__(self, aggregation: CompValue):
  54. super(Counter, self).__init__(aggregation)
  55. self.value = 0
  56. if self.expr == "*":
  57. # cannot eval "*" => always use the full row
  58. # type error: Cannot assign to a method
  59. self.eval_row = self.eval_full_row # type: ignore[assignment]
  60. def update(self, row: FrozenBindings, aggregator: Aggregator) -> None:
  61. try:
  62. val = self.eval_row(row)
  63. except NotBoundError:
  64. # skip UNDEF
  65. return
  66. self.value += 1
  67. if self.distinct:
  68. self.seen.add(val)
  69. def get_value(self) -> Literal:
  70. return Literal(self.value)
  71. def eval_row(self, row: FrozenBindings) -> Identifier:
  72. return _eval(self.expr, row)
  73. def eval_full_row(self, row: FrozenBindings) -> FrozenBindings:
  74. return row
  75. def use_row(self, row: FrozenBindings) -> bool:
  76. try:
  77. return self.eval_row(row) not in self.seen
  78. except NotBoundError:
  79. # happens when counting zero optional nodes. See issue #2229
  80. return False
  81. @overload
  82. def type_safe_numbers(*args: int) -> Tuple[int]: ...
  83. @overload
  84. def type_safe_numbers(
  85. *args: Union[Decimal, float, int]
  86. ) -> Tuple[Union[float, int]]: ...
  87. def type_safe_numbers(*args: Union[Decimal, float, int]) -> Iterable[Union[float, int]]:
  88. if any(isinstance(arg, float) for arg in args) and any(
  89. isinstance(arg, Decimal) for arg in args
  90. ):
  91. return map(float, args)
  92. # type error: Incompatible return value type (got "Tuple[Union[Decimal, float, int], ...]", expected "Iterable[Union[float, int]]")
  93. # NOTE on type error: if args contains a Decimal it will nopt get here.
  94. return args # type: ignore[return-value]
  95. class Sum(Accumulator):
  96. def __init__(self, aggregation: CompValue):
  97. super(Sum, self).__init__(aggregation)
  98. self.value = 0
  99. self.datatype: Optional[str] = None
  100. def update(self, row: FrozenBindings, aggregator: Aggregator) -> None:
  101. try:
  102. value = _eval(self.expr, row)
  103. dt = self.datatype
  104. if dt is None:
  105. dt = value.datatype
  106. else:
  107. # type error: Argument 1 to "type_promotion" has incompatible type "str"; expected "URIRef"
  108. dt = type_promotion(dt, value.datatype) # type: ignore[arg-type]
  109. self.datatype = dt
  110. self.value = sum(type_safe_numbers(self.value, numeric(value)))
  111. if self.distinct:
  112. self.seen.add(value)
  113. except NotBoundError:
  114. # skip UNDEF
  115. pass
  116. def get_value(self) -> Literal:
  117. return Literal(self.value, datatype=self.datatype)
  118. class Average(Accumulator):
  119. def __init__(self, aggregation: CompValue):
  120. super(Average, self).__init__(aggregation)
  121. self.counter = 0
  122. self.sum = 0
  123. self.datatype: Optional[str] = None
  124. def update(self, row: FrozenBindings, aggregator: Aggregator) -> None:
  125. try:
  126. value = _eval(self.expr, row)
  127. dt = self.datatype
  128. self.sum = sum(type_safe_numbers(self.sum, numeric(value)))
  129. if dt is None:
  130. dt = value.datatype
  131. else:
  132. # type error: Argument 1 to "type_promotion" has incompatible type "str"; expected "URIRef"
  133. dt = type_promotion(dt, value.datatype) # type: ignore[arg-type]
  134. self.datatype = dt
  135. if self.distinct:
  136. self.seen.add(value)
  137. self.counter += 1
  138. # skip UNDEF or BNode => SPARQLTypeError
  139. except NotBoundError:
  140. pass
  141. except SPARQLTypeError:
  142. pass
  143. def get_value(self) -> Literal:
  144. if self.counter == 0:
  145. return Literal(0)
  146. if self.datatype in (XSD.float, XSD.double):
  147. return Literal(self.sum / self.counter)
  148. else:
  149. return Literal(Decimal(self.sum) / Decimal(self.counter))
  150. class Extremum(Accumulator):
  151. """abstract base class for Minimum and Maximum"""
  152. def __init__(self, aggregation: CompValue):
  153. self.compare: Callable[[Any, Any], Any]
  154. super(Extremum, self).__init__(aggregation)
  155. self.value: Any = None
  156. # DISTINCT would not change the value for MIN or MAX
  157. # type error: Cannot assign to a method
  158. self.use_row = self.dont_care # type: ignore[method-assign]
  159. def set_value(self, bindings: MutableMapping[Variable, Identifier]) -> None:
  160. if self.value is not None:
  161. # simply do not set if self.value is still None
  162. bindings[self.var] = Literal(self.value)
  163. def update(self, row: FrozenBindings, aggregator: Aggregator) -> None:
  164. try:
  165. if self.value is None:
  166. self.value = _eval(self.expr, row)
  167. else:
  168. # self.compare is implemented by Minimum/Maximum
  169. self.value = self.compare(self.value, _eval(self.expr, row))
  170. # skip UNDEF or BNode => SPARQLTypeError
  171. except NotBoundError:
  172. pass
  173. except SPARQLTypeError:
  174. pass
  175. _ValueT = TypeVar("_ValueT", Variable, BNode, URIRef, Literal)
  176. class Minimum(Extremum):
  177. def compare(self, val1: _ValueT, val2: _ValueT) -> _ValueT:
  178. return min(val1, val2, key=_val)
  179. class Maximum(Extremum):
  180. def compare(self, val1: _ValueT, val2: _ValueT) -> _ValueT:
  181. return max(val1, val2, key=_val)
  182. class Sample(Accumulator):
  183. """takes the first eligible value"""
  184. def __init__(self, aggregation):
  185. super(Sample, self).__init__(aggregation)
  186. # DISTINCT would not change the value
  187. # type error: Cannot assign to a method
  188. self.use_row = self.dont_care # type: ignore[method-assign]
  189. def update(self, row: FrozenBindings, aggregator: Aggregator) -> None:
  190. try:
  191. # set the value now
  192. aggregator.bindings[self.var] = _eval(self.expr, row)
  193. # and skip this accumulator for future rows
  194. del aggregator.accumulators[self.var]
  195. except NotBoundError:
  196. pass
  197. def get_value(self) -> None:
  198. # set None if no value was set
  199. return None
  200. class GroupConcat(Accumulator):
  201. value: List[Literal]
  202. def __init__(self, aggregation: CompValue):
  203. super(GroupConcat, self).__init__(aggregation)
  204. # only GROUPCONCAT needs to have a list as accumulator
  205. self.value = []
  206. if aggregation.separator is None:
  207. self.separator = " "
  208. else:
  209. self.separator = aggregation.separator
  210. def update(self, row: FrozenBindings, aggregator: Aggregator) -> None:
  211. try:
  212. value = _eval(self.expr, row)
  213. # skip UNDEF
  214. if isinstance(value, NotBoundError):
  215. return
  216. self.value.append(value)
  217. if self.distinct:
  218. self.seen.add(value)
  219. # skip UNDEF
  220. # NOTE: It seems like this is not the way undefined values occur, they
  221. # come through not as exceptions but as values. This is left here
  222. # however as it may occur in some cases.
  223. # TODO: Consider removing this.
  224. except NotBoundError:
  225. pass
  226. def get_value(self) -> Literal:
  227. return Literal(self.separator.join(str(v) for v in self.value))
  228. class Aggregator:
  229. """combines different Accumulator objects"""
  230. accumulator_classes = {
  231. "Aggregate_Count": Counter,
  232. "Aggregate_Sample": Sample,
  233. "Aggregate_Sum": Sum,
  234. "Aggregate_Avg": Average,
  235. "Aggregate_Min": Minimum,
  236. "Aggregate_Max": Maximum,
  237. "Aggregate_GroupConcat": GroupConcat,
  238. }
  239. def __init__(self, aggregations: List[CompValue]):
  240. self.bindings: Dict[Variable, Identifier] = {}
  241. self.accumulators: Dict[str, Accumulator] = {}
  242. for a in aggregations:
  243. accumulator_class = self.accumulator_classes.get(a.name)
  244. if accumulator_class is None:
  245. raise Exception("Unknown aggregate function " + a.name)
  246. self.accumulators[a.res] = accumulator_class(a)
  247. def update(self, row: FrozenBindings) -> None:
  248. """update all own accumulators"""
  249. # SAMPLE accumulators may delete themselves
  250. # => iterate over list not generator
  251. for acc in list(self.accumulators.values()):
  252. if acc.use_row(row):
  253. acc.update(row, self)
  254. def get_bindings(self) -> Mapping[Variable, Identifier]:
  255. """calculate and set last values"""
  256. for acc in self.accumulators.values():
  257. acc.set_value(self.bindings)
  258. return self.bindings