Mercurial > libervia-pubsub
comparison sat_pubsub/pgsql_storage.py @ 451:0e6e176cb572
pgsql: fix items order:
Due to a misinterpretation, items where returned in wrong order when RSM was used.
Pubsub vanilla `max_items` is reversing the items order to get last items, but when this
attribute is not used (RSM being used or not), the items must be in chronological order.
This patch fixes it, so RSM returns oldest items by defaut, and empty `<before/>` must be
used to get last page (and thus most recent items).
Items are always finally ordered chronologically inside the returned page (default by
`update` time, but this can be modified when a specific order is requested).
Reference field is also fixed to use `updated` instead of `item_id` when item are ordered
by `update`.
author | Goffi <goffi@goffi.org> |
---|---|
date | Sun, 06 Jun 2021 19:42:57 +0200 |
parents | 96059231d58f |
children | 7f1394bb96db |
comparison
equal
deleted
inserted
replaced
450:96059231d58f | 451:0e6e176cb572 |
---|---|
1 #!/usr/bin/env python3 | 1 #!/usr/bin/env python3 |
2 #-*- coding: utf-8 -*- | |
3 | 2 |
4 # Copyright (c) 2012-2021 Jérôme Poisson | 3 # Copyright (c) 2012-2021 Jérôme Poisson |
5 # Copyright (c) 2013-2016 Adrien Cossa | 4 # Copyright (c) 2013-2016 Adrien Cossa |
6 # Copyright (c) 2003-2011 Ralph Meijer | 5 # Copyright (c) 2003-2011 Ralph Meijer |
7 | 6 |
880 class LeafNode(Node): | 879 class LeafNode(Node): |
881 | 880 |
882 | 881 |
883 nodeType = 'leaf' | 882 nodeType = 'leaf' |
884 | 883 |
885 def getOrderBy(self, ext_data, direction='DESC'): | 884 def getOrderBy(self, ext_data, direction='ASC'): |
886 """Return ORDER BY clause corresponding to Order By key in ext_data | 885 """Return ORDER BY clause corresponding to Order By key in ext_data |
887 | 886 |
888 @param ext_data (dict): extra data as used in getItems | 887 @param ext_data (dict): extra data as used in getItems |
889 @param direction (unicode): ORDER BY direction (ASC or DESC) | 888 @param direction (unicode): ORDER BY direction (ASC or DESC) |
890 @return (unicode): ORDER BY clause to use | 889 @return (unicode): ORDER BY clause to use |
1137 else: | 1136 else: |
1138 log.msg("WARNING: unknown filter: {}".format(filter_.encode('utf-8'))) | 1137 log.msg("WARNING: unknown filter: {}".format(filter_.encode('utf-8'))) |
1139 | 1138 |
1140 query.extend(query_filters) | 1139 query.extend(query_filters) |
1141 | 1140 |
1142 return self.getOrderBy(ext_data) | 1141 return self.getOrderBy(ext_data, direction="ASC") |
1143 | 1142 |
1144 def _getItems(self, cursor, authorized_groups, unrestricted, maxItems, ext_data, ids_only): | 1143 def _getItems(self, cursor, authorized_groups, unrestricted, maxItems, ext_data, ids_only): |
1145 self._checkNodeExists(cursor) | 1144 self._checkNodeExists(cursor) |
1146 | 1145 |
1147 if maxItems == 0: | 1146 if maxItems == 0: |
1154 query = ["SELECT item"] | 1153 query = ["SELECT item"] |
1155 else: | 1154 else: |
1156 query = ["SELECT data::text,items.access_model,item_id,created,updated"] | 1155 query = ["SELECT data::text,items.access_model,item_id,created,updated"] |
1157 | 1156 |
1158 query_order = self._appendSourcesAndFilters(query, args, authorized_groups, unrestricted, ext_data) | 1157 query_order = self._appendSourcesAndFilters(query, args, authorized_groups, unrestricted, ext_data) |
1158 if query_order.startswith("ORDER BY updated"): | |
1159 ref_field = "updated" | |
1160 else: | |
1161 ref_field = "item_id" | |
1159 | 1162 |
1160 if 'rsm' in ext_data: | 1163 if 'rsm' in ext_data: |
1161 rsm = ext_data['rsm'] | 1164 rsm = ext_data['rsm'] |
1162 maxItems = rsm.max | 1165 maxItems = rsm.max |
1163 if rsm.index is not None: | 1166 if rsm.index is not None: |
1176 query.append("AND item_id<=%s") | 1179 query.append("AND item_id<=%s") |
1177 args.append(item_id) | 1180 args.append(item_id) |
1178 elif rsm.before is not None: | 1181 elif rsm.before is not None: |
1179 if rsm.before != '': | 1182 if rsm.before != '': |
1180 query.append( | 1183 query.append( |
1181 "AND item_id>(SELECT item_id FROM items WHERE node_id=%s AND " | 1184 f"AND {ref_field}<(SELECT {ref_field} FROM items WHERE " |
1182 "item=%s LIMIT 1)" | 1185 "node_id=%s AND item=%s LIMIT 1)" |
1183 ) | 1186 ) |
1184 args.extend((self.nodeDbId, rsm.before)) | 1187 args.extend((self.nodeDbId, rsm.before)) |
1185 if maxItems is not None: | 1188 # we need to reverse order in a first query to get the right |
1186 # if we have maxItems (i.e. a limit), we need to reverse order | 1189 # items |
1187 # in a first query to get the right items | 1190 query.insert(0,"SELECT * from (") |
1188 query.insert(0,"SELECT * from (") | 1191 query.append(self.getOrderBy(ext_data, direction='DESC')) |
1189 query.append(self.getOrderBy(ext_data, direction='ASC')) | 1192 query.append("LIMIT %s) as x") |
1190 query.append("LIMIT %s) as x") | 1193 if maxItems is None: |
1194 # we default to 10 max when we request last page | |
1195 log.msg( | |
1196 "WARNING: getting last page with RSM without max, " | |
1197 "we default to max=10" | |
1198 ) | |
1199 args.append(10) | |
1200 else: | |
1191 args.append(maxItems) | 1201 args.append(maxItems) |
1192 elif rsm.after: | 1202 elif rsm.after: |
1193 query.append( | 1203 query.append( |
1194 "AND item_id<(SELECT item_id FROM items WHERE node_id=%s AND item=%s " | 1204 f"AND {ref_field}>(SELECT {ref_field} FROM items WHERE node_id=%s " |
1195 "LIMIT 1)" | 1205 "AND item=%s LIMIT 1)" |
1196 ) | 1206 ) |
1197 args.extend((self.nodeDbId, rsm.after)) | 1207 args.extend((self.nodeDbId, rsm.after)) |
1208 else: | |
1209 rsm = False | |
1198 | 1210 |
1199 query.append(query_order) | 1211 query.append(query_order) |
1200 | 1212 |
1201 if maxItems is not None: | 1213 if maxItems is not None: |
1202 query.append("LIMIT %s") | 1214 if rsm == False: |
1203 args.append(maxItems) | 1215 # if we have legacy maxItems (i.e. not RSM max), we need to |
1216 # reverse order (maxItems get last items, cf. | |
1217 # https://xmpp.org/extensions/xep-0060.html#subscriber-retrieve-requestrecent) | |
1218 # in a first query to get the right items | |
1219 query.insert(0,"SELECT * from (") | |
1220 query.insert(len(query)-1, self.getOrderBy(ext_data, direction='DESC')) | |
1221 query.insert(len(query)-1, "LIMIT %s) as x") | |
1222 args.append(maxItems) | |
1223 else: | |
1224 query.append("LIMIT %s") | |
1225 args.append(maxItems) | |
1204 | 1226 |
1205 cursor.execute(' '.join(query), args) | 1227 cursor.execute(' '.join(query), args) |
1206 | 1228 |
1207 result = cursor.fetchall() | 1229 result = cursor.fetchall() |
1208 if unrestricted and not ids_only: | 1230 if unrestricted and not ids_only: |