Matrix/synapse
Matrix/synapse
1
0
Fork 0
mirror of https://github.com/element-hq/synapse synced 2024-09-17 18:55:10 +00:00
Code Issues Projects Releases Packages Wiki Activity

More comments

Browse source
This commit is contained in:
Eric Eastwood 2023-05-18 02:31:12 -05:00
parent 333fc51e6c
commit 17aeee764d
1 changed files with 22 additions and 10 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
synapse/storage/databases/state/bg_updates.py
View file

@ -90,7 +90,15 @@ class StateGroupBackgroundUpdateStore(SQLBaseStore):
state_filter: Optional[StateFilter] = None, state_filter: Optional[StateFilter] = None,
) -> Mapping[int, StateMap[str]]: ) -> Mapping[int, StateMap[str]]:
""" """
TODO Given a number of state groups, fetch the latest state for each group.
Args:
txn: The transaction object.
groups: The given state groups that you want to fetch the latest state for.
state_filter: The state filter to apply the state we fetch state from the database.
Returns:
Map from state_group to a StateMap at that point.
""" """
state_filter = state_filter or StateFilter.all() state_filter = state_filter or StateFilter.all()
@ -102,13 +110,11 @@ class StateGroupBackgroundUpdateStore(SQLBaseStore):
# a temporary hack until we can add the right indices in # a temporary hack until we can add the right indices in
txn.execute("SET LOCAL enable_seqscan=off") txn.execute("SET LOCAL enable_seqscan=off")
# The below query walks the state_group tree so that the "state" # The query below walks the state_group tree so that the "state"
# table includes all state_groups in the tree. It then joins # table includes all state_groups in the tree. It then joins
# against `state_groups_state` to fetch the latest state. # against `state_groups_state` to fetch the latest state.
# It assumes that previous state groups are always numerically # It assumes that previous state groups are always numerically
# lesser. # lesser.
# This may return multiple rows per (type, state_key), but last_value
# should be the same.
sql = """ sql = """
WITH RECURSIVE sgs(state_group, state_group_reached) AS ( WITH RECURSIVE sgs(state_group, state_group_reached) AS (
VALUES(?::bigint, NULL::bigint) VALUES(?::bigint, NULL::bigint)
@ -214,13 +220,19 @@ class StateGroupBackgroundUpdateStore(SQLBaseStore):
ORDER BY type, state_key, state_group DESC ORDER BY type, state_key, state_group DESC
""" """
# We can sort from smallest to largest state_group and re-use the work from # We can sort from least to greatest state_group and re-use the work from a
# the small state_group for a larger one if we see that the edge chain links # lesser state_group for a greater one if we see that the edge chain links
# up. # up.
#
# What this means in practice is that if we fetch the latest state for
# `state_group = 20`, and then we want `state_group = 30`, it will traverse
# down the edge chain to `20`, see that we linked up to `20` and bail out
# early and re-use the work we did for `20`.
sorted_groups = sorted(groups) sorted_groups = sorted(groups)
state_groups_we_have_already_fetched: Set[int] = set( state_groups_we_have_already_fetched: Set[int] = set(
# We default to `[-1]` just to fill in the query with something # We default to `[-1]` just to fill in the query with something that
# that will have no effect but not bork our query when it would be empty otherwise # will have no effect but not bork our query when it would be empty
# otherwise
[-1] [-1]
) )
for group in sorted_groups: for group in sorted_groups:
@ -244,7 +256,7 @@ class StateGroupBackgroundUpdateStore(SQLBaseStore):
# The first row is always our special `state_group_reached` row which # The first row is always our special `state_group_reached` row which
# tells us if we linked up to any other existing state_group that we # tells us if we linked up to any other existing state_group that we
# already fetched and if so, which one we linked up to (see the `UNION # already fetched and if so, which one we linked up to (see the `UNION
# ALL` above) # ALL` above which drives this special row)
first_row = txn.fetchone() first_row = txn.fetchone()
if first_row: if first_row:
_, _, _, state_group_reached = first_row _, _, _, state_group_reached = first_row
@ -255,7 +267,7 @@ class StateGroupBackgroundUpdateStore(SQLBaseStore):
key = (intern_string(typ), intern_string(state_key)) key = (intern_string(typ), intern_string(state_key))
partial_state_map_for_state_group[key] = event_id partial_state_map_for_state_group[key] = event_id
# If we see a state group edge link to a previous state_group that we # If we see a state_group edge link to a previous state_group that we
# already fetched from the database, link up the base state to the # already fetched from the database, link up the base state to the
# partial state we retrieved from the database to build on top of. # partial state we retrieved from the database to build on top of.
if state_group_reached in results: if state_group_reached in results: