Mercurial > libervia-pubsub

comparison sat_pubsub/pgsql_storage.py @ 451:0e6e176cb572

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
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: