aminghadersohi commented on code in PR #44009:
URL: https://github.com/apache/superset/pull/44009#discussion_r4002402898
##########
superset/versioning/changes/listener.py:
##########
@@ -395,7 +396,51 @@ def _persist_buffered_records(
incr_capture_error("bulk_insert")
-def register_change_record_listener() -> None: # noqa: C901
+def finalize_change_records(session: Session) -> None:
+ """Build and persist the transaction's change records at commit time.
+
+ Module-level (rather than a closure inside the registration function)
+ so the capture write path can be exercised directly by unit tests
+ against an isolated session; it depends only on the session and the
+ module helpers, never on the registered entity classes.
+ """
+ if session.in_nested_transaction() or session.info.get(_FINALIZING_KEY):
+ return
+
+ session.info[_FINALIZING_KEY] = True
+ # Measures the FINALIZE stage only: the timer starts after the flush,
+ # which excludes the transaction's own write cost but also excludes
+ # capture_initial_states' per-entity pre-state SELECTs — and runs
+ # through every capture step and early return. Every commit on the
+ # session emits a sample, including commits touching no versioned
+ # entity, because the whole-listener overhead is exactly what the
+ # kill-switch removes; a flush that raises emits nothing.
+ start: float | None = None
+ try:
+ session.flush()
+ start = perf_counter()
+ initial_states: dict[tuple[str, int], tuple[Any, dict[str, Any]]] = (
+ session.info.get(_INITIAL_STATES_KEY, {})
+ )
+ buffer = _build_scalar_buffer(initial_states)
+
+ tx_id = _current_transaction_id(session)
Review Comment:
Confirmed by probe, not just by reading: with the Continuum uow lookup
patched to raise, `session.commit()` propagated the `RuntimeError` and **0 rows
persisted**; the same session with the lookup returning `None` committed fine.
The invariant is actually broken, not merely unguarded.
```suggestion
try:
tx_id = _current_transaction_id(session)
except Exception: # pylint: disable=broad-except
logger.exception("version_changes: transaction lookup failed")
incr_capture_error("transaction_lookup")
return
```
##########
superset/versioning/changes/listener.py:
##########
@@ -395,7 +396,51 @@ def _persist_buffered_records(
incr_capture_error("bulk_insert")
-def register_change_record_listener() -> None: # noqa: C901
+def finalize_change_records(session: Session) -> None:
+ """Build and persist the transaction's change records at commit time.
+
+ Module-level (rather than a closure inside the registration function)
+ so the capture write path can be exercised directly by unit tests
+ against an isolated session; it depends only on the session and the
+ module helpers, never on the registered entity classes.
+ """
+ if session.in_nested_transaction() or session.info.get(_FINALIZING_KEY):
+ return
+
+ session.info[_FINALIZING_KEY] = True
+ # The latency series measures CAPTURE overhead only: the timer starts
+ # after the transaction's own final flush — a cost that exists with
+ # versioning disabled and must not be charged to capture — and runs
Review Comment:
Taken verbatim in `ec56800a` — the comment now states both exclusions
accurately, so this thread is resolved from our side. The unreconciled
alerting-purpose half is live in @fitzee's thread on line 421.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
