From 74bf26afb901e2985568eb44d51a7478f6201a34 Mon Sep 17 00:00:00 2001 From: Hannah Bast Date: Tue, 28 Jul 2026 17:19:54 +0200 Subject: [PATCH 1/4] Gate out-of-order events also across batches The gating from #299 (an out-of-order delete must not cancel a causally newer insert of the same triple, and vice versa) only worked WITHIN a batch, because the `insert_triples` / `delete_triples` dicts are reset for each batch. When tailing the live stream, a batch spans only a few seconds, while the stream can reorder events by minutes, so an out-of-order event can arrive in a later batch than the event it causally precedes, ungated. Observed live (found by `qlever check-sync-with-wikidata`): a sitelink was moved from a duplicate entity to the right one; the delete for the duplicate (rev 2523832444) arrived 2:23 minutes after the insert for the right entity (rev 2523832445) and stripped the article triples, which are identical for both entities except for `schema:about`. Keep a history of the causally newest event per triple ACROSS batches (new option `--triple-history-minutes`, default 30, 0 disables), gate inserts and deletes against it with the same tie semantics as the per-batch gate, and prune the history by event date, also within a batch (a batch can span hours of stream time during catch-up). --- src/qlever/commands/update_wikidata.py | 130 ++++++++++++++++++++++++- 1 file changed, 129 insertions(+), 1 deletion(-) diff --git a/src/qlever/commands/update_wikidata.py b/src/qlever/commands/update_wikidata.py index b7d9630b..42c217a7 100644 --- a/src/qlever/commands/update_wikidata.py +++ b/src/qlever/commands/update_wikidata.py @@ -7,7 +7,7 @@ import re import signal import time -from datetime import datetime, timezone +from datetime import datetime, timedelta, timezone from enum import Enum, auto from pathlib import Path from threading import Event @@ -239,6 +239,16 @@ def additional_arguments(self, subparser) -> None: help="Process exactly this many messages and then exit " "(default: no bound on the number of messages)", ) + subparser.add_argument( + "--triple-history-minutes", + type=int, + default=30, + help="Gate inserts and deletes against the causally newest " + "event seen for the same triple within this many minutes of " + "stream time, ACROSS batches; the stream can reorder events " + "by minutes, so gating only within a batch is not enough " + "(0 disables the history)", + ) subparser.add_argument( "--verbose", choices=["no", "yes"], @@ -390,6 +400,26 @@ def determine_next_cached_update( return cached_file_name, batch_size + def prune_triple_history( + self, triple_history, latest_event_date, history_minutes + ): + """ + Return the given triple history without the entries that are older + than `history_minutes` before the given latest event date (the + event dates are ISO strings and compare lexicographically). + """ + if not triple_history or latest_event_date is None: + return triple_history + cutoff_date = ( + datetime.strptime(latest_event_date, "%Y-%m-%dT%H:%M:%SZ") + - timedelta(minutes=history_minutes) + ).strftime("%Y-%m-%dT%H:%M:%SZ") + return { + triple: history + for triple, history in triple_history.items() + if history[2] >= cutoff_date + } + def execute(self, args) -> bool: # Resolve the four args that depend on `--wikimedia-commons`. A # `None` value means the user did not pass that arg, so fill in @@ -512,6 +542,23 @@ def execute(self, args) -> bool: log.error(f"Error determining offset from stream: {e}") return False + # History of the causal-order key of the chronologically newest + # event seen for each triple, kept ACROSS batches: maps the triple + # to `(rev_key, is_insert, event_date)`. The per-batch + # `insert_triples` / `delete_triples` below gate out-of-order + # events WITHIN a batch, but the stream can reorder events by + # minutes, while batches span only seconds when tailing the live + # stream. Without this history, a delete that arrives after the + # causally newer insert of the same triple was applied in an + # earlier batch deletes that triple (observed live: a sitelink + # moved from a duplicate entity to the right one, where the + # delete for the duplicate arrived two minutes late and stripped + # the article triples of the right entity). Entries older than + # `--triple-history-minutes` of stream time are pruned at each + # batch start. + triple_history: dict[str, tuple[tuple[int, int], bool, str]] = {} + latest_event_date = None + # Initialize all the statistics variables. batch_count = 0 total_num_messages = 0 @@ -700,6 +747,14 @@ def execute(self, args) -> bool: insert_triples: dict[str, tuple[int, int]] = {} delete_triples: dict[str, tuple[int, int]] = {} + # Prune old entries from the triple history. + if args.triple_history_minutes > 0: + triple_history = self.prune_triple_history( + triple_history, + latest_event_date, + args.triple_history_minutes, + ) + # Check if we can use a cached SPARQL query file use_cached_file = False cached_file_name = None @@ -746,6 +801,11 @@ def execute(self, args) -> bool: # Get the date (rounded *down* to seconds). date = meta.get("dt") date = re.sub(r"\.\d*Z$", "Z", date) + if ( + latest_event_date is None + or date > latest_event_date + ): + latest_event_date = date # Get the other relevant fields from the message. entity_id = event_data.get("entity_id") @@ -892,6 +952,35 @@ def node_to_sparql(node: rdflib.term.Node) -> str: ) for s, p, o in graph: triple = f"{s.n3()} {p.n3()} {node_to_sparql(o)}" + # Cross-batch gate (see `triple_history`): discard + # this delete if a causally newer (or equally new) + # insert of the same triple was seen, possibly in an + # earlier batch that has been applied already. The tie + # semantics match the per-batch gate below. + history = ( + triple_history.get(triple) + if args.triple_history_minutes + > 0 + else None + ) + if ( + history is not None + and history[1] + and not rev_key > history[0] + ): + continue + if ( + args.triple_history_minutes > 0 + and ( + history is None + or rev_key > history[0] + ) + ): + triple_history[triple] = ( + rev_key, + False, + date, + ) # NOTE: In case there was a previous `insert` of that # triple, it is safe to remove that `insert`, but not # the `delete` (in case the triple is contained in the @@ -942,6 +1031,34 @@ def node_to_sparql(node: rdflib.term.Node) -> str: ) for s, p, o in graph: triple = f"{s.n3()} {p.n3()} {node_to_sparql(o)}" + # Cross-batch gate (see `triple_history`): discard + # this insert if a causally strictly newer delete of + # the same triple was seen, possibly in an earlier + # batch that has been applied already. + history = ( + triple_history.get(triple) + if args.triple_history_minutes + > 0 + else None + ) + if ( + history is not None + and not history[1] + and not rev_key >= history[0] + ): + continue + if ( + args.triple_history_minutes > 0 + and ( + history is None + or rev_key >= history[0] + ) + ): + triple_history[triple] = ( + rev_key, + True, + date, + ) # NOTE: In case there was a previous `delete` of that # triple, it is safe to remove that `delete`, but not # the `insert` (in case the triple is not contained in @@ -980,6 +1097,17 @@ def node_to_sparql(node: rdflib.term.Node) -> str: # Message was successfully processed, update batch tracking current_batch_size += 1 + if ( + args.triple_history_minutes > 0 + and current_batch_size % 10000 == 0 + ): + # A batch can span hours of stream time during + # catch-up, so also prune within a batch. + triple_history = self.prune_triple_history( + triple_history, + latest_event_date, + args.triple_history_minutes, + ) total_num_messages += 1 pbar_update_frequency = 100 if (current_batch_size % pbar_update_frequency) == 0: From 0c5b925e55770f919d03633df92df8813128e4be Mon Sep 17 00:00:00 2001 From: Hannah Bast Date: Mon, 10 Aug 2026 19:35:35 +0200 Subject: [PATCH 2/4] Address review feedback Replace the "gate" terminology in the option help text and the comments (an event that is superseded by a causally newer event is now said to be ignored). Factor the two near-identical cross-batch checks (one for deletes, one for inserts) into a single helper check_and_update_triple_history, which also removes the deeply indented duplicated blocks. Add type annotations to prune_triple_history and a NOTE that cached batches contribute nothing to the triple history (Copilot comments). --- src/qlever/commands/update_wikidata.py | 122 +++++++++++++------------ 1 file changed, 66 insertions(+), 56 deletions(-) diff --git a/src/qlever/commands/update_wikidata.py b/src/qlever/commands/update_wikidata.py index 42c217a7..21d5401e 100644 --- a/src/qlever/commands/update_wikidata.py +++ b/src/qlever/commands/update_wikidata.py @@ -243,11 +243,11 @@ def additional_arguments(self, subparser) -> None: "--triple-history-minutes", type=int, default=30, - help="Gate inserts and deletes against the causally newest " - "event seen for the same triple within this many minutes of " - "stream time, ACROSS batches; the stream can reorder events " - "by minutes, so gating only within a batch is not enough " - "(0 disables the history)", + help="Remember the causally newest event for each triple for " + "this many minutes of stream time, ACROSS batches, and ignore " + "inserts and deletes that are superseded by such an event; the " + "stream can reorder events by minutes, so checking only within " + "a batch is not enough (0 disables the history)", ) subparser.add_argument( "--verbose", @@ -401,8 +401,11 @@ def determine_next_cached_update( return cached_file_name, batch_size def prune_triple_history( - self, triple_history, latest_event_date, history_minutes - ): + self, + triple_history: dict[str, tuple[tuple[int, int], bool, str]], + latest_event_date: str | None, + history_minutes: int, + ) -> dict[str, tuple[tuple[int, int], bool, str]]: """ Return the given triple history without the entries that are older than `history_minutes` before the given latest event date (the @@ -420,6 +423,37 @@ def prune_triple_history( if history[2] >= cutoff_date } + def check_and_update_triple_history( + self, + triple_history: dict[str, tuple[tuple[int, int], bool, str]], + triple: str, + rev_key: tuple[int, int], + is_insert: bool, + event_date: str, + ) -> bool: + """ + Check the given event against the history of the causally newest + event per triple (see `triple_history` in `execute`) and return + whether the event should be applied. An insert is superseded by a + causally strictly newer delete of the same triple, a delete by a + causally newer or equally new insert (an insert wins a tie, a + delete does not, matching the per-batch tie semantics). If the + event is the causally newest one for its triple so far, it is + recorded in the history (which is modified in place). + """ + history = triple_history.get(triple) + is_newest = history is None or ( + rev_key >= history[0] if is_insert else rev_key > history[0] + ) + if is_newest: + triple_history[triple] = (rev_key, is_insert, event_date) + return True + + # An event that is not the causally newest one is superseded only + # by an event of the other kind (a delete by an insert and vice + # versa); a duplicate of the same kind is harmless. + return history[1] == is_insert + def execute(self, args) -> bool: # Resolve the four args that depend on `--wikimedia-commons`. A # `None` value means the user did not pass that arg, so fill in @@ -545,7 +579,7 @@ def execute(self, args) -> bool: # History of the causal-order key of the chronologically newest # event seen for each triple, kept ACROSS batches: maps the triple # to `(rev_key, is_insert, event_date)`. The per-batch - # `insert_triples` / `delete_triples` below gate out-of-order + # `insert_triples` / `delete_triples` below catch out-of-order # events WITHIN a batch, but the stream can reorder events by # minutes, while batches span only seconds when tailing the live # stream. Without this history, a delete that arrives after the @@ -742,8 +776,8 @@ def execute(self, args) -> bool: # DELETE and the target's ADD for a transferred article URL share # the same `dt` and the target often arrives first). Tracking the # `rev_id` per triple makes the cross-set "remove from the other - # side" step gated by causal order, so the chronologically last - # event for a given triple wins regardless of arrival order. + # side" step respect the causal order, so the chronologically + # last event for a given triple wins regardless of arrival order. insert_triples: dict[str, tuple[int, int]] = {} delete_triples: dict[str, tuple[int, int]] = {} @@ -756,6 +790,13 @@ def execute(self, args) -> bool: ) # Check if we can use a cached SPARQL query file + # + # NOTE: A cached batch is applied without processing its events, + # so it contributes nothing to `triple_history`. For events in + # such a batch, the protection against out-of-order events is + # therefore only as good as it was before the introduction of + # `triple_history` (namely, within the batch that produced the + # cached file). use_cached_file = False cached_file_name = None if ( @@ -952,35 +993,19 @@ def node_to_sparql(node: rdflib.term.Node) -> str: ) for s, p, o in graph: triple = f"{s.n3()} {p.n3()} {node_to_sparql(o)}" - # Cross-batch gate (see `triple_history`): discard - # this delete if a causally newer (or equally new) - # insert of the same triple was seen, possibly in an - # earlier batch that has been applied already. The tie - # semantics match the per-batch gate below. - history = ( - triple_history.get(triple) - if args.triple_history_minutes - > 0 - else None - ) - if ( - history is not None - and history[1] - and not rev_key > history[0] - ): - continue + # Cross-batch check, see + # `check_and_update_triple_history`. if ( args.triple_history_minutes > 0 - and ( - history is None - or rev_key > history[0] - ) - ): - triple_history[triple] = ( + and not self.check_and_update_triple_history( + triple_history, + triple, rev_key, False, date, ) + ): + continue # NOTE: In case there was a previous `insert` of that # triple, it is safe to remove that `insert`, but not # the `delete` (in case the triple is contained in the @@ -1031,38 +1056,23 @@ def node_to_sparql(node: rdflib.term.Node) -> str: ) for s, p, o in graph: triple = f"{s.n3()} {p.n3()} {node_to_sparql(o)}" - # Cross-batch gate (see `triple_history`): discard - # this insert if a causally strictly newer delete of - # the same triple was seen, possibly in an earlier - # batch that has been applied already. - history = ( - triple_history.get(triple) - if args.triple_history_minutes - > 0 - else None - ) - if ( - history is not None - and not history[1] - and not rev_key >= history[0] - ): - continue + # Cross-batch check, see + # `check_and_update_triple_history`. if ( args.triple_history_minutes > 0 - and ( - history is None - or rev_key >= history[0] - ) - ): - triple_history[triple] = ( + and not self.check_and_update_triple_history( + triple_history, + triple, rev_key, True, date, ) + ): + continue # NOTE: In case there was a previous `delete` of that # triple, it is safe to remove that `delete`, but not # the `insert` (in case the triple is not contained in - # the original data). Use `>=` on the cross-set gate + # the original data). Use `>=` on the cross-set check # so that a same-event delete-then-add (deletes are # processed first in the per-event loop) lets the # add win, matching the previous within-event From a42cab5732918307a2a0b3dac7507467c4b8e1d2 Mon Sep 17 00:00:00 2001 From: Hannah Bast Date: Mon, 10 Aug 2026 19:56:49 +0200 Subject: [PATCH 3/4] Amortize the pruning of the triple history A prune is linear in the size of the history, which matters when tailing the live stream, where a prune would otherwise run once per batch, that is, every few seconds. Now a prune only actually happens when the history has at least doubled in size since the last prune (with a floor of 10,000 entries), which makes the total pruning cost linear in the number of insertions. Keeping entries longer than the window is harmless for correctness, the window is only a bound on the memory usage. --- src/qlever/commands/update_wikidata.py | 18 +++++++++++++++++- 1 file changed, 17 insertions(+), 1 deletion(-) diff --git a/src/qlever/commands/update_wikidata.py b/src/qlever/commands/update_wikidata.py index 21d5401e..c5aed4e1 100644 --- a/src/qlever/commands/update_wikidata.py +++ b/src/qlever/commands/update_wikidata.py @@ -129,6 +129,9 @@ def __init__(self): self.ctrl_c_pressed = Event() # Set to `True` when finished. self.finished = False + # Size of the triple history right after the last prune (see + # `prune_triple_history`). + self.triple_history_size_after_last_prune = 0 def description(self) -> str: return "Update from given SSE stream" @@ -410,18 +413,31 @@ def prune_triple_history( Return the given triple history without the entries that are older than `history_minutes` before the given latest event date (the event dates are ISO strings and compare lexicographically). + + A prune is linear in the size of the history, so it only actually + happens when the history has at least doubled in size since the + last prune; the total pruning cost is then linear in the total + number of insertions, no matter how often this is called. Keeping + entries longer than `history_minutes` is harmless for correctness, + the window is only a bound on the memory usage. """ if not triple_history or latest_event_date is None: return triple_history + if len(triple_history) < max( + 2 * self.triple_history_size_after_last_prune, 10_000 + ): + return triple_history cutoff_date = ( datetime.strptime(latest_event_date, "%Y-%m-%dT%H:%M:%SZ") - timedelta(minutes=history_minutes) ).strftime("%Y-%m-%dT%H:%M:%SZ") - return { + triple_history = { triple: history for triple, history in triple_history.items() if history[2] >= cutoff_date } + self.triple_history_size_after_last_prune = len(triple_history) + return triple_history def check_and_update_triple_history( self, From ece59af7bb1284b6f73a79700564ecf330dfa8f2 Mon Sep 17 00:00:00 2001 From: Hannah Bast Date: Mon, 10 Aug 2026 20:10:45 +0200 Subject: [PATCH 4/4] Pass is_insert and event_date as keyword arguments and hoist the use-triple-history condition into a variable --- src/qlever/commands/update_wikidata.py | 17 +++++++++-------- 1 file changed, 9 insertions(+), 8 deletions(-) diff --git a/src/qlever/commands/update_wikidata.py b/src/qlever/commands/update_wikidata.py index c5aed4e1..db03f1ca 100644 --- a/src/qlever/commands/update_wikidata.py +++ b/src/qlever/commands/update_wikidata.py @@ -608,6 +608,7 @@ def execute(self, args) -> bool: # batch start. triple_history: dict[str, tuple[tuple[int, int], bool, str]] = {} latest_event_date = None + use_triple_history = args.triple_history_minutes > 0 # Initialize all the statistics variables. batch_count = 0 @@ -798,7 +799,7 @@ def execute(self, args) -> bool: delete_triples: dict[str, tuple[int, int]] = {} # Prune old entries from the triple history. - if args.triple_history_minutes > 0: + if use_triple_history: triple_history = self.prune_triple_history( triple_history, latest_event_date, @@ -1012,13 +1013,13 @@ def node_to_sparql(node: rdflib.term.Node) -> str: # Cross-batch check, see # `check_and_update_triple_history`. if ( - args.triple_history_minutes > 0 + use_triple_history and not self.check_and_update_triple_history( triple_history, triple, rev_key, - False, - date, + is_insert=False, + event_date=date, ) ): continue @@ -1075,13 +1076,13 @@ def node_to_sparql(node: rdflib.term.Node) -> str: # Cross-batch check, see # `check_and_update_triple_history`. if ( - args.triple_history_minutes > 0 + use_triple_history and not self.check_and_update_triple_history( triple_history, triple, rev_key, - True, - date, + is_insert=True, + event_date=date, ) ): continue @@ -1124,7 +1125,7 @@ def node_to_sparql(node: rdflib.term.Node) -> str: # Message was successfully processed, update batch tracking current_batch_size += 1 if ( - args.triple_history_minutes > 0 + use_triple_history and current_batch_size % 10000 == 0 ): # A batch can span hours of stream time during