mikebridge commented on code in PR #41076: URL: https://github.com/apache/superset/pull/41076#discussion_r3575000318
########## superset/versioning/activity/kinds.py: ########## @@ -0,0 +1,210 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Kind translation tables, shared types, and the shadow-model loader. + +The activity-view module operates in two "kind" namespaces — the +table-stored value (``"chart"`` / ``"dashboard"`` / ``"dataset"``) that +``version_changes.entity_kind`` carries, and the Python class-name +form (``"Slice"`` / ``"Dashboard"`` / ``"SqlaTable"``) used internally +for dispatch and returned in the DTO's ``entity_kind`` field. The four +mappings here translate between them. Adjacent kind-keyed dicts live +here too: the per-kind human-readable label, the user-facing +lowercase form, and the 404 exception class. + +The :func:`load_live_model` helper exists in the same module +because each lookup is keyed on the same set of class names — keeping +it adjacent to the mappings makes the kind-translation surface +discoverable at a glance. +""" + +from __future__ import annotations + +from collections.abc import Iterable, Iterator +from dataclasses import dataclass + +from flask_appbuilder import Model + +from superset.commands.chart.exceptions import ChartNotFoundError +from superset.commands.dashboard.exceptions import DashboardNotFoundError +from superset.commands.dataset.exceptions import DatasetNotFoundError +from superset.versioning.changes import ENTITY_KIND_BY_CLASS_NAME + +# Max entity ids per ``IN (...)`` clause across the activity-view queries. +# Bounds the bind-parameter count under SQLite's ``SQLITE_MAX_VARIABLE_NUMBER`` +# floor (default 999 in many builds); Postgres/MySQL accept the full list but +# the chunk is dialect-agnostic. 500 leaves headroom for the other bound +# params on each statement. +ENTITY_ID_CHUNK_SIZE = 500 + + +def chunked_ids( + ids: Iterable[int], size: int = ENTITY_ID_CHUNK_SIZE +) -> Iterator[list[int]]: + """Yield *ids* in fixed-size lists (final chunk may be smaller). Shared by + the activity-view query helpers so every ``IN (...)`` stays under the + SQLite bind-variable floor.""" + items = list(ids) + for i in range(0, len(items), size): + yield items[i : i + size] + + +# ---- Kind translation ----------------------------------------------------- + +# ``version_changes.entity_kind`` stores the friendly downstream-tooling +# value (``"chart"``, ``"dashboard"``, ``"dataset"``) per the +# ``ENTITY_KIND_BY_CLASS_NAME``. The activity-view DTO returns the +# Python class name instead (``"Slice"``, ``"Dashboard"``, +# ``"SqlaTable"``) so the contract aligns with ``__class__.__name__`` +# (the ``ActivityRecord`` DTO). Translate at the boundary. +TABLE_KIND_TO_API: dict[str, str] = { + table_kind: class_name + for class_name, table_kind in ENTITY_KIND_BY_CLASS_NAME.items() +} +API_KIND_TO_TABLE: dict[str, str] = dict(ENTITY_KIND_BY_CLASS_NAME) + +# Human-readable label for summary headlines +# ("Dataset updated: Sales Transactions"). Keyed by the internal API kind +# (Python class name; matches ``model_cls.__name__``). +API_KIND_LABEL: dict[str, str] = { + "Dashboard": "Dashboard", + "Slice": "Chart", + "SqlaTable": "Dataset", +} + +# User-facing lowercase rendering of the kind. This is what appears in +# the JSON response's ``entity_kind`` field and the +# ``ActivityRecordSchema.entity_kind`` enum. Internal code keeps the +# Python class-name form because it matches ``model_cls.__name__`` and is +# convenient for dispatch — translation happens at serialization time +# only, in :func:`render.apply_record_decoration`. +USER_FACING_KIND: dict[str, str] = { + "Dashboard": "dashboard", + "Slice": "chart", + "SqlaTable": "dataset", +} + +# 404 exception class per API kind. Each accepts a string positional arg +# (the path-entity UUID) that gets formatted into the exception message. +NOT_FOUND_EXC: dict[str, type[Exception]] = { + "Dashboard": DashboardNotFoundError, + "Slice": ChartNotFoundError, + "SqlaTable": DatasetNotFoundError, +} + +# Per-API-kind (model class name, display column) used by +# ``_resolve_names_for_kind`` to read the user-facing entity name from +# the shadow table valid at a given transaction. +NAME_COLUMN: dict[str, tuple[str, str]] = { + "Dashboard": ("Dashboard", "dashboard_title"), + "Slice": ("Slice", "slice_name"), + "SqlaTable": ("SqlaTable", "table_name"), +} + + +# ---- Types ---------------------------------------------------------------- + + +@dataclass(frozen=True) +class Window: + """A validity window in Continuum transaction-id space, half-open as + ``[start_tx, end_tx)``. + + Immutable and equal-by-attributes — two windows with the same + ``start_tx`` / ``end_tx`` are interchangeable. Constructor rejects + ``end_tx <= start_tx``. ``end_tx = None`` means "open ended + (current)" and acts as positive infinity throughout the helpers. + + Helper methods (``contains`` / ``intersect`` / ``merges_with``) + live on the type so callers don't re-implement the half-open + predicate inline. Previously a ``tuple[int, int | None]`` alias; + promoted to a dataclass so a function accepting a ``Window`` can't + silently accept any other 2-tuple and so the constructor enforces + the half-open invariant. + """ + + start_tx: int + end_tx: int | None + + def __post_init__(self) -> None: + if self.end_tx is not None and self.end_tx <= self.start_tx: + raise ValueError( + f"Window end_tx must be > start_tx; " + f"got [{self.start_tx}, {self.end_tx})" + ) + + def contains(self, tx_id: int) -> bool: + """``True`` iff *tx_id* falls inside this half-open interval.""" + return self.start_tx <= tx_id and (self.end_tx is None or tx_id < self.end_tx) + + def intersect(self, other: Window) -> Window | None: + """Return the clipped overlap of this window with *other*, or + ``None`` when they are disjoint. ``end_tx = None`` acts as + positive infinity on either side.""" + start = max(self.start_tx, other.start_tx) + end: int | None + if self.end_tx is None: + end = other.end_tx + elif other.end_tx is None: + end = self.end_tx + else: + end = min(self.end_tx, other.end_tx) + if end is not None and end <= start: + return None + return Window(start, end) + + def merges_with(self, other: Window) -> bool: + """``True`` iff *self* and *other* overlap or touch (so their + union is one contiguous window). Assumes the caller has placed + them in start-ascending order.""" + if self.end_tx is None: + # self extends to +∞; everything past it merges in. + return True + return other.start_tx <= self.end_tx + + +#: A related-entity scope row: ``(api_kind, entity_id, [windows])``. +#: ``api_kind`` is the DTO-facing kind (``"Slice"``, etc.), not the +#: table-stored kind. Modelled as a tuple alias — the +#: ``(api_kind, entity_id)`` pair is logically a key with the window +#: list as its value, so a dict shape may fit better than a flat +#: dataclass. +EntityWindows = tuple[str, int, list[Window]] + + +def load_live_model(model_name: str) -> type[Model]: + """Inline-import a live model class by name. Deferred until call + time because the versioning package is initialised before all model + mappers are configured (same idiom used throughout + :mod:`superset.versioning.changes`). + + Returns the LIVE model class (``Dashboard`` / ``Slice`` / + ``SqlaTable``), not a Continuum shadow class — callers derive the + shadow table separately via ``version_class(...)`` where needed.""" + # pylint: disable=import-outside-toplevel + if model_name == "Dashboard": + from superset.models.dashboard import Dashboard + + return Dashboard + if model_name == "Slice": + from superset.models.slice import Slice + + return Slice + if model_name == "SqlaTable": + from superset.connectors.sqla.models import SqlaTable + + return SqlaTable + raise LookupError(f"No shadow class registered for {model_name!r}") Review Comment: Addressed by subsequent commits; the flagged code is no longer current after the rebase onto master. Resolving as outdated. ########## tests/integration_tests/versioning/activity_view_tests.py: ########## @@ -0,0 +1,1250 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Integration tests for the cross-entity activity-view API (sc-107283). + +US1 — dashboard activity stream: ``GET /api/v1/dashboard/<uuid>/activity/``. +Tests for US2 (chart activity) and US3 (dataset activity) come in later +phases. + +Per spec T053 / sc-103156 T062, every test that mutates a fixture entity +wraps the test body in ``try``/``finally`` with +``metadata_db.session.rollback()`` in the ``finally``. The rationale is +documented in the spec — Continuum captures dirty mappers during +autoflush, so leaving an instrumented attribute dirty pollutes +downstream tests via the shadow tables. +""" + +from __future__ import annotations + +from typing import Any +from uuid import uuid4 + +import pytest + +from superset.connectors.sqla.models import SqlaTable +from superset.extensions import db +from superset.models.dashboard import Dashboard +from superset.models.slice import Slice +from superset.utils import json as _json +from tests.integration_tests.base_tests import SupersetTestCase +from tests.integration_tests.constants import ( + ADMIN_USERNAME, + ALPHA_USERNAME, + GAMMA_USERNAME, +) +from tests.integration_tests.fixtures.birth_names_dashboard import ( # noqa: F401 + load_birth_names_dashboard_with_slices, + load_birth_names_data, +) + + +def _get_birth_names_dataset() -> SqlaTable: + return ( + db.session.query(SqlaTable) + .filter(SqlaTable.table_name == "birth_names") + .first() + ) + + +def _persist_fixture_state() -> None: + """Force the fixture's pending INSERTs to commit so subsequent edits + produce *new* version rows instead of being batched into the + creation transaction. Mirrors the same helper in + ``tests/integration_tests/dashboards/version_history_tests.py``. + """ + db.session.commit() + + +def _get_birth_names_dashboard() -> Dashboard: + return ( + db.session.query(Dashboard) + .filter(Dashboard.dashboard_title == "USA Births Names") + .first() + ) + + +class TestDashboardActivityView(SupersetTestCase): + """T017–T026 — ``GET /api/v1/dashboard/<uuid>/activity/`` (US1).""" + + @pytest.fixture(autouse=True) + def _load_data(self, load_birth_names_dashboard_with_slices): # noqa: PT004, F811 + pass + + def _activity(self, dashboard_uuid: str, **query: Any) -> Any: + return self.client.get( + f"/api/v1/dashboard/{dashboard_uuid}/activity/", + query_string=query, + ) + + # ---- 4xx boundary cases ---- + + def test_activity_returns_404_for_unknown_uuid(self) -> None: + """AV-009: unknown path entity → 404.""" + self.login(ADMIN_USERNAME) + rv = self._activity("00000000-0000-0000-0000-000000000000") + assert rv.status_code == 404 + + def test_activity_returns_400_for_invalid_uuid(self) -> None: + """A malformed UUID is rejected by the endpoint, not by Werkzeug.""" + self.login(ADMIN_USERNAME) + rv = self._activity("not-a-uuid") + assert rv.status_code == 400 + + def test_activity_returns_400_for_invalid_include(self) -> None: + _persist_fixture_state() + dashboard = _get_birth_names_dashboard() + assert dashboard is not None + self.login(ADMIN_USERNAME) + rv = self._activity(str(dashboard.uuid), include="sibling") + assert rv.status_code == 400 + + def test_activity_returns_400_for_invalid_since(self) -> None: + _persist_fixture_state() + dashboard = _get_birth_names_dashboard() + assert dashboard is not None + self.login(ADMIN_USERNAME) + rv = self._activity(str(dashboard.uuid), since="yesterday") + assert rv.status_code == 400 + + def test_activity_allows_read_non_owner(self) -> None: + """Activity is a read endpoint: a non-owner with read access (Alpha, + which carries broad read + datasource access) can read a dashboard's + activity stream — ``raise_for_access(dashboard=)`` does not reject — + so the endpoint returns 200. Visibility of *related* rows is filtered + separately, inside the activity layer.""" + _persist_fixture_state() + dashboard = _get_birth_names_dashboard() + assert dashboard is not None + dashboard_uuid = str(dashboard.uuid) + + self.login(ALPHA_USERNAME) + rv = self._activity(dashboard_uuid) + assert rv.status_code == 200 + + def test_visibility_filter_silently_drops_inaccessible_related(self) -> None: + """AV-008 security control: a related record whose entity the caller + cannot read is *silently* dropped — absent from the result, no + placeholder, no contribution to count. Exercises the real + enforcement point (``filter_records_by_visibility`` → + ``DashboardAccessFilter``) with a restricted Gamma principal. + + Setup uses two dashboards (no datasource needed): one owned by + Gamma (readable), and an admin-owned one Gamma may not read. + """ + from superset import security_manager + from superset.subjects.utils import get_user_subject + from superset.versioning.activity.visibility import ( + filter_records_by_visibility, + ) + + admin = security_manager.find_user(ADMIN_USERNAME) + gamma = security_manager.find_user(GAMMA_USERNAME) + # Access control is subject-based (``editors``) since the Subject + # work replaced ``owners``; grant read via each user's Subject. + visible = Dashboard( + dashboard_title=f"vis-probe-owned {uuid4().hex[:8]}", + slug=f"vis-owned-{uuid4().hex[:8]}", + published=False, + editors=[get_user_subject(gamma.id)], + ) + hidden = Dashboard( + dashboard_title=f"vis-probe-hidden {uuid4().hex[:8]}", + slug=f"vis-hidden-{uuid4().hex[:8]}", + published=False, + editors=[get_user_subject(admin.id)], + ) + db.session.add_all([visible, hidden]) + db.session.commit() + visible_id, hidden_id = visible.id, hidden.id + try: + records = [ + {"entity_kind": "dashboard", "entity_id": visible_id}, + {"entity_kind": "dashboard", "entity_id": hidden_id}, + ] + self.login(GAMMA_USERNAME) + filtered = filter_records_by_visibility(records) + kept_ids = {r["entity_id"] for r in filtered} + + assert visible_id in kept_ids, ( + "gamma-owned dashboard must stay visible to Gamma" + ) + assert hidden_id not in kept_ids, ( + "unpublished admin-owned dashboard must be dropped for Gamma" + ) + # Silent: the dropped record leaves nothing behind (no placeholder). + assert len(filtered) == 1 + finally: + db.session.rollback() + for did in (visible_id, hidden_id): + obj = db.session.query(Dashboard).filter(Dashboard.id == did).first() + if obj is not None: + db.session.delete(obj) + db.session.commit() + + # ---- 200 happy paths ---- + + def test_activity_returns_200_with_envelope_shape(self) -> None: + """Smoke test: the endpoint returns the documented envelope shape + (``result`` list + ``count`` integer) even when the dashboard has + no activity yet.""" + _persist_fixture_state() + dashboard = _get_birth_names_dashboard() + assert dashboard is not None + dashboard_uuid = str(dashboard.uuid) + + self.login(ADMIN_USERNAME) + rv = self._activity(dashboard_uuid) + assert rv.status_code == 200 + body = _json.loads(rv.data.decode("utf-8")) + assert "result" in body + assert "count" in body + assert isinstance(body["result"], list) + assert isinstance(body["count"], int) + + def test_activity_includes_chart_edit_as_related(self) -> None: + """T018 / AS-1 of US1: editing a chart on the dashboard surfaces + the chart-edit record with ``entity_kind=Slice`` and + ``source=related``.""" + _persist_fixture_state() + dashboard = _get_birth_names_dashboard() + assert dashboard is not None + dashboard_uuid = str(dashboard.uuid) + chart_on_dashboard = next(iter(dashboard.slices), None) + assert chart_on_dashboard is not None + chart_id = chart_on_dashboard.id + original_name = chart_on_dashboard.slice_name + + try: + chart_on_dashboard.slice_name = f"{original_name} (edited)" + db.session.commit() + + self.login(ADMIN_USERNAME) + rv = self._activity(dashboard_uuid) + assert rv.status_code == 200 + body = _json.loads(rv.data.decode("utf-8")) + related = [ + r + for r in body["result"] + if r["entity_kind"] == "chart" and r["source"] == "related" + ] + assert related, ( + "Expected at least one Slice/related record from the chart " + "edit; got: " + f"{[(r['entity_kind'], r['source']) for r in body['result']]}" + ) + # Spot-check the carry-through of denormalized fields + sample = related[0] + assert sample["entity_uuid"] is not None + assert "transaction_id" in sample + assert "issued_at" in sample + finally: + db.session.rollback() + chart = db.session.query(Slice).filter(Slice.id == chart_id).one() + chart.slice_name = original_name + db.session.commit() + + def test_activity_include_self_excludes_related(self) -> None: + """T023 / AV-016: ``?include=self`` filters out related records.""" + _persist_fixture_state() + dashboard = _get_birth_names_dashboard() + assert dashboard is not None + dashboard_uuid = str(dashboard.uuid) + chart_on_dashboard = next(iter(dashboard.slices), None) + assert chart_on_dashboard is not None + chart_id = chart_on_dashboard.id + original_name = chart_on_dashboard.slice_name + + try: + chart_on_dashboard.slice_name = f"{original_name} (edited self)" + db.session.commit() + + self.login(ADMIN_USERNAME) + rv = self._activity(dashboard_uuid, include="self") + assert rv.status_code == 200 + body = _json.loads(rv.data.decode("utf-8")) + for record in body["result"]: + assert record["source"] == "self", ( + f"include=self leaked a non-self record: {record}" + ) + assert record["entity_kind"] == "dashboard" + finally: + db.session.rollback() + chart = db.session.query(Slice).filter(Slice.id == chart_id).one() + chart.slice_name = original_name + db.session.commit() + + def test_activity_include_related_excludes_self(self) -> None: + """T024 / AV-016: ``?include=related`` returns only related records.""" + _persist_fixture_state() + dashboard = _get_birth_names_dashboard() + assert dashboard is not None + dashboard_uuid = str(dashboard.uuid) + original_title = dashboard.dashboard_title + dashboard_id = dashboard.id + + try: + # Edit the dashboard's own field so we have a self record to + # filter out, and edit a chart on it so we have a related + # record to keep. + dashboard.dashboard_title = f"{original_title} (edited dash)" + db.session.commit() + chart_on_dashboard = next(iter(dashboard.slices), None) + assert chart_on_dashboard is not None + chart_id = chart_on_dashboard.id + chart_original_name = chart_on_dashboard.slice_name + chart_on_dashboard.slice_name = f"{chart_original_name} (edited chart)" + db.session.commit() + + self.login(ADMIN_USERNAME) + rv = self._activity(dashboard_uuid, include="related") + assert rv.status_code == 200 + body = _json.loads(rv.data.decode("utf-8")) + for record in body["result"]: + assert record["source"] == "related", ( + f"include=related leaked a self record: {record}" + ) + assert record["entity_kind"] != "dashboard" + finally: + db.session.rollback() + dashboard = ( + db.session.query(Dashboard).filter(Dashboard.id == dashboard_id).one() + ) + dashboard.dashboard_title = original_title + chart = db.session.query(Slice).filter(Slice.id == chart_id).one() + chart.slice_name = chart_original_name + db.session.commit() Review Comment: Addressed in the preceding review-fix commit: cleanup variables are initialized before the `try` and guarded before restoration. ########## tests/integration_tests/versioning/perf_validation_tests.py: ########## @@ -0,0 +1,416 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""T044 — Performance validation for entity version history. + +Skipped by default. Run on demand: + + SUPERSET_PERF_VALIDATION=1 pytest \ + tests/integration_tests/versioning/perf_validation_tests.py -v -s + +Measures the three success criteria defined in the spec: + + * SC-002: version list endpoint responds in under 1 second + * SC-004: save path p95 overhead under 50 ms with Continuum tracking + on vs. off (FR-014) + +(SC-003, the restore-endpoint benchmark, lands with the restore endpoint +in a later PR.) + +The test prints a summary table suitable for pasting into the PR +description. It also asserts each target so regressions fail loudly +when the harness is re-run. +""" + +from __future__ import annotations + +import os +import statistics +import time +from typing import Any + +import pytest +import sqlalchemy as sa +from sqlalchemy_continuum import version_class, versioning_manager + +from superset.extensions import db +from superset.models.slice import Slice +from tests.integration_tests.base_tests import SupersetTestCase +from tests.integration_tests.constants import ADMIN_USERNAME +from tests.integration_tests.fixtures.birth_names_dashboard import ( # noqa: F401 + load_birth_names_dashboard_with_slices, + load_birth_names_data, +) + +SKIP_REASON = "Performance validation is manual. Set SUPERSET_PERF_VALIDATION=1 to run." + +# Thresholds from spec.md §Success Criteria. +LIST_ENDPOINT_MAX_MS = 1000 # SC-002 +SAVE_OVERHEAD_P95_MAX_MS = 50 # SC-004 + +# Activity-view thresholds (sc-107283 §Success Criteria). +ACTIVITY_ENDPOINT_P95_MAX_MS = 1500 # SC-AV-001 + + +def _save_chart_once(chart: Slice, suffix: str) -> None: + """One ORM-level save path, mimicking what ChartDAO.update does.""" + chart.slice_name = f"{chart.slice_name[:64]}_{suffix}" + db.session.commit() + + +def _timings_ms(seconds: list[float]) -> dict[str, float]: + ms = sorted(s * 1000.0 for s in seconds) + return { + "p50": statistics.median(ms), + "p95": ms[int(len(ms) * 0.95) - 1] if len(ms) >= 20 else max(ms), + "max": max(ms), + "n": len(ms), + } + + [email protected]( + not os.environ.get("SUPERSET_PERF_VALIDATION"), + reason=SKIP_REASON, +) +class PerfValidationTests(SupersetTestCase): + """Runs only when SUPERSET_PERF_VALIDATION=1 is set.""" + + @pytest.fixture(autouse=True) + def _load_data(self, load_birth_names_dashboard_with_slices: Any) -> None: # noqa: F811, PT004 + pass + + def _seed_chart_with_n_versions(self, n: int) -> Slice: + """Save a chart N times to produce N version rows.""" + chart = db.session.query(Slice).first() + assert chart is not None, "birth_names fixture should provide charts" + + for i in range(n): + _save_chart_once(chart, f"v{i}") + db.session.commit() + return chart + + def test_sc002_list_endpoint_under_1s(self) -> None: + """SC-002: list endpoint responds in under 1 second.""" + self.login(ADMIN_USERNAME) + + # Generate enough versions to exercise the retention-capped state. + chart = self._seed_chart_with_n_versions(24) + chart_uuid = str(chart.uuid) + url = f"/api/v1/chart/{chart_uuid}/versions/" + + # Warm up the endpoint once (JIT caching, mapper configuration, etc.) + self.client.get(url) + + timings: list[float] = [] + for _ in range(10): + t0 = time.perf_counter() + response = self.client.get(url) + timings.append(time.perf_counter() - t0) + assert response.status_code == 200 + + stats = _timings_ms(timings) + print( + f"\n[SC-002] GET /versions/ (24 versions) " + f"p50={stats['p50']:.1f}ms p95={stats['p95']:.1f}ms " + f"max={stats['max']:.1f}ms n={stats['n']}" + ) + assert stats["p95"] < LIST_ENDPOINT_MAX_MS, ( + f"SC-002 failed: list endpoint p95 {stats['p95']:.1f}ms " + f">= {LIST_ENDPOINT_MAX_MS}ms" + ) + + # SC-003 (restore endpoint under 3s) intentionally omitted here: the + # version-restore route (POST /<uuid>/versions/<version_uuid>/restore) + # ships in a later PR, so there is nothing to time at this point in the + # stack. Re-add this benchmark alongside the restore endpoint. + + def test_sc004_save_overhead_under_50ms(self) -> None: + """SC-004: save path p95 overhead under 50ms (FR-014). + + Toggling Continuum on and off mid-process corrupts its internal + ``units_of_work`` state and is not a reliable measurement. Instead + this test directly measures the wall-clock time spent inside the + four session-level listeners Continuum attaches to + ``sa.orm.session.Session`` — ``before_flush``, ``after_flush``, + ``after_commit``, ``after_rollback`` — plus Superset's own + baseline / snapshot / retention-prune listeners (attached to + ``db.session``). The cumulative listener time per save is the + marginal overhead version capture adds over a save with + versioning removed entirely, because without these listeners + the ORM would not execute any of that code. + + The approach: + 1. Wrap each known listener with a timing proxy that adds its + wall-clock time to a per-save accumulator. + 2. Save the same chart N times, recording each save's + accumulator value. + 3. Compute p50 / p95 of the per-save overhead. + + This matches the measurement intent of SC-004 (how much does + versioning cost per save) without the fragility of toggling + Continuum mid-test. + """ + self.login(ADMIN_USERNAME) + + chart = db.session.query(Slice).first() + assert chart is not None + + # Per-save accumulator incremented by the wrapped listeners. + acc = [0.0] + + def wrap_listener(original: Any) -> Any: + def wrapper(*args: Any, **kwargs: Any) -> Any: + t0 = time.perf_counter() + try: + return original(*args, **kwargs) + finally: + acc[0] += time.perf_counter() - t0 + + wrapper.__wrapped__ = original # type: ignore[attr-defined] + return wrapper + + # Instrument Continuum's four session listeners by detaching the + # bound method, wrapping, and re-attaching under a single-use + # listener handle we can cleanly remove on teardown. + session_target = sa.orm.session.Session + attached: list[tuple[str, Any]] = [] + for event_name, listener in list(versioning_manager.session_listeners.items()): + sa.event.remove(session_target, event_name, listener) + wrapped = wrap_listener(listener) + sa.event.listen(session_target, event_name, wrapped) + attached.append((event_name, wrapped)) Review Comment: Fixed in 8b19744b85. The performance gates now use full save latency, which includes both Continuum listeners and Superset listeners attached to `db.session`; the wrapped-listener timing is diagnostic only. ########## superset/versioning/activity/orchestrator.py: ########## @@ -0,0 +1,434 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Top-level orchestrator + query-param parsing + observability. + +This is the public entry point for the activity-view read path. One +function — :func:`get_activity` — dispatches on the path entity's +model class to assemble the cross-entity activity stream: + +1. ``resolve_path_entity`` (queries.py) — resolve UUID → live entity. +2. ``resolve_scope`` (scope.py) — build the related-entity window list. +3. ``fetch_change_records`` (queries.py) — pull rows from + ``version_changes`` joined with ``version_transaction`` and ``ab_user``. +4. ``filter_records_by_visibility`` (visibility.py) — silent + drop of records the requester can't read. +5. ``apply_entity_name_denormalization`` (queries.py) — resolve entity names + from the shadow row valid at each record's transaction_id. +6. ``apply_record_decoration`` (render.py) — synthesize the ActivityRecord + DTO fields and strip internal-only columns. +7. Paginate in Python over the post-filter list. + +Parameter parsing for the REST endpoints lives here too — +:func:`parse_activity_query_params` is called by the three +``/activity/`` endpoint handlers before they call ``get_activity``. +Same for the observability instrumentation: ``_phase_timer`` and +``_emit_request_shape_attributes`` emit the per-phase timing and +request-shape metrics, on the same prefix the cross-coupling test pins. +""" + +from __future__ import annotations + +import contextlib +from collections.abc import Iterator +from datetime import datetime, timezone +from typing import Any +from uuid import UUID + +from flask import Response +from flask_appbuilder import Model + +from superset.utils import json +from superset.versioning.activity.kinds import EntityWindows +from superset.versioning.activity.queries import ( + apply_entity_name_denormalization, + fetch_change_records, + first_tracked_tx, + mark_first_tracked_saves, + resolve_path_entity, +) +from superset.versioning.activity.render import apply_record_decoration +from superset.versioning.activity.scope import resolve_scope +from superset.versioning.activity.visibility import filter_records_by_visibility +from superset.versioning.api_helpers import ( + PathEntityResponseError, + resolve_endpoint_path_entity, +) + +_DEFAULT_PAGE_SIZE = 25 +_MAX_PAGE_SIZE = 200 +_VALID_INCLUDE_VALUES: frozenset[str] = frozenset({"self", "related", "all"}) + + +# Upper bound on the ``q`` search string. The search is a substring scan over +# the (already-capped) materialized record set, so this is a cheap-DoS guard, +# not a correctness limit. +_MAX_Q_LENGTH = 1024 + + +class ActivityParamsError(ValueError): + """Raised by :func:`parse_activity_query_params` when a query param is + malformed. The endpoint catches this and maps to ``response_400``; + no other callers should depend on the exception type.""" + + +def parse_activity_query_params(args: Any) -> dict[str, Any]: + """Parse the ``since`` / ``until`` / ``include`` / ``page`` / ``page_size`` + query parameters into the kwargs ``get_activity`` accepts. + + Raises :class:`ActivityParamsError` (subclass of ``ValueError``) when + a parameter is malformed. Shared across the three endpoint families + (dashboards, charts, datasets) so the parsing and 400-messaging stay + consistent. + """ + params: dict[str, Any] = { + "include": _parse_include(args.get("include", "all")), + "page": _parse_page(args.get("page", "0")), + "page_size": _parse_page_size(args.get("page_size")), + } + if (since := _parse_optional_iso(args.get("since"), name="since")) is not None: + params["since"] = since + if (until := _parse_optional_iso(args.get("until"), name="until")) is not None: + params["until"] = until + if q := (args.get("q") or "").strip(): + if len(q) > _MAX_Q_LENGTH: + raise ActivityParamsError(f"'q' must be at most {_MAX_Q_LENGTH} characters") + params["q"] = q + return params + + +def _parse_optional_iso(raw: str | None, *, name: str) -> datetime | None: + """Parse a missing-or-ISO-datetime field; ``None`` for missing, + ``ActivityParamsError`` for malformed.""" + if not raw: + return None + parsed = _parse_iso_datetime(raw) + if parsed is None: + raise ActivityParamsError(f"Invalid {name!r} datetime: {raw!r}") + return parsed + + +def _parse_include(value: str) -> str: + if value not in _VALID_INCLUDE_VALUES: + raise ActivityParamsError( + f"Invalid 'include' value: {value!r}; " + f"must be one of {sorted(_VALID_INCLUDE_VALUES)}" + ) + return value + + +def _parse_page(raw: str) -> int: + try: + value = int(raw) + except (TypeError, ValueError) as exc: + raise ActivityParamsError(f"Invalid 'page' value: {raw!r}") from exc + if value < 0: + raise ActivityParamsError("Invalid 'page' value: must be >= 0") + return value + + +def _parse_page_size(raw: str | None) -> int: + """``page_size`` honours the default when missing, raises when invalid, + and silently clamps to ``_MAX_PAGE_SIZE`` (so ``?page_size=500`` + returns 200 records instead of a 400).""" + if raw is None: + return _DEFAULT_PAGE_SIZE + try: + value = int(raw) + except (TypeError, ValueError) as exc: + raise ActivityParamsError(f"Invalid 'page_size' value: {raw!r}") from exc + if value < 1: + raise ActivityParamsError("Invalid 'page_size' value: must be >= 1") + return min(value, _MAX_PAGE_SIZE) + + +def _parse_iso_datetime(value: str) -> datetime | None: + """Parse an ISO-8601 datetime string. Tolerates the trailing ``Z`` + suffix that Python <3.11 ``fromisoformat`` rejects, and normalises any + timezone-aware result to naive UTC. + + The ``since`` / ``until`` filters bind directly against + ``version_transaction.issued_at``, which is ``sa.DateTime()`` — a + timezone-*naive* column (UTC by convention). Binding a tz-aware value + against it shifts the comparison by the session offset on PostgreSQL + (and raises on some drivers), so collapse aware inputs to naive UTC + here. Naive inputs pass through unchanged (already treated as UTC). + """ + candidate = value[:-1] + "+00:00" if value.endswith("Z") else value + try: + parsed = datetime.fromisoformat(candidate) + except ValueError: + return None + if parsed.tzinfo is not None: + parsed = parsed.astimezone(timezone.utc).replace(tzinfo=None) + return parsed + + +def _record_matches(record: dict[str, Any], q: str) -> bool: + """Case-insensitive substring match for the ``q`` search filter, + over the human-meaningful surfaces of a decorated activity record: + ``summary``, ``entity_name``, ``kind``, the joined ``path`` segments, + and the JSON form of ``from_value`` / ``to_value`` (JSON, not Python + ``str()``: the client searches the serialized text it renders, so + ``false`` / ``null`` / double-quoted keys must match — and falsy + values like ``False`` / ``0`` must not collapse to unsearchable + empty strings). + """ + + def _value_text(value: Any) -> str: + if value is None: + return "" + try: + return json.dumps(value) + except (TypeError, ValueError): + return str(value) + + needle = q.lower() + haystacks = ( + record.get("summary") or "", + record.get("entity_name") or "", + record.get("kind") or "", + " ".join(str(seg) for seg in (record.get("path") or [])), + _value_text(record.get("from_value")), + _value_text(record.get("to_value")), + ) + return any(needle in h.lower() for h in haystacks) + + +def get_activity( + model_cls: type[Model], + entity_uuid: UUID, + *, + since: datetime | None = None, + until: datetime | None = None, + include: str = "all", + q: str | None = None, + page: int = 0, + page_size: int = _DEFAULT_PAGE_SIZE, + resolved_entity: Any | None = None, +) -> tuple[list[dict[str, Any]], int, bool]: + """Cross-entity activity stream for one path entity. + + Single polymorphic entry point. Dispatches on *model_cls* to + assemble the path entity's self records plus the transitive related- + entity records (charts attached to a dashboard, datasets a chart + pointed at, etc.). + + Returns ``(records, total_count, truncated)``. ``truncated`` is + ``True`` when the per-kind fetch ceiling + (``queries._MAX_FETCHED_RECORDS``) bit for any kind — the stream is + clamped to a clean time cut and older records exist beyond it, so + ``count`` is a floor, not the absolute total. + + *resolved_entity*, when supplied, is the already-resolved live path + entity (the endpoint resolves it once via ``resolve_endpoint_path_entity`` + for the ``raise_for_access`` gate); passing it here skips a second + identical ``find_active_by_uuid`` lookup and the TOCTOU window between + them. The count is post-visibility (silent visibility filter), + post-include-filter, and — when ``q`` is supplied — post- + search-filter (``count`` reflects the matches, the contract the + server-side search exists to provide), not just the size of the + returned slice — clients paginate by passing ``page`` forward until + ``page * page_size >= count``. + + Raises ``DashboardNotFoundError`` / ``ChartNotFoundError`` / + ``DatasetNotFoundError`` when the path entity doesn't exist. + """ + if resolved_entity is not None: + path_entity, path_id = resolved_entity, resolved_entity.id + else: + path_entity, path_id = resolve_path_entity(model_cls, entity_uuid) + path_kind = model_cls.__name__ + kind_key = path_kind.lower() # "dashboard" / "slice" / "sqlatable" + + # Lower-bound the self window at the entity's first tracked transaction + # (matched on (id, uuid)) so a reused integer id can't inherit a + # hard-deleted predecessor's history. + self_start_tx = first_tracked_tx(model_cls, path_id, path_entity.uuid) + with _phase_timer(kind_key, "relationship_resolution_ms"): + entity_windows = resolve_scope(path_kind, path_id, include, self_start_tx) + if not entity_windows: + _emit_request_shape_attributes( + kind_key, + include=include, + has_since_filter=since is not None, + page_size=page_size, + record_count=0, + entity_windows=[], + path_kind=path_kind, + path_id=path_id, + ) + return [], 0, False + + # Visibility filter runs before decoration: it needs the raw + # ``entity_id`` column (which decoration strips), and dropping + # invisible records early means we don't pay for name lookup + + # tombstone probes + impact counts on records the requester + # can't see (the silent-filter contract). + with _phase_timer(kind_key, "fetch_ms"): + records, truncated = fetch_change_records(entity_windows, since, until) + with _phase_timer(kind_key, "visibility_filter_ms"): + records = filter_records_by_visibility(records) + with _phase_timer(kind_key, "denormalize_ms"): + apply_entity_name_denormalization(records) + # Runs post-visibility (fewer entities to probe) and pre- + # decoration (needs the raw table-form entity_kind/entity_id + # that decoration rewrites). + mark_first_tracked_saves(records) + with _phase_timer(kind_key, "decorate_ms"): + apply_record_decoration(records, path_kind, path_id) + + # Server-side search (the panel's client-side + # search only covers loaded pages). Applied post-decoration so the + # synthesized ``summary`` / ``entity_name`` participate, and pre- + # count so pagination paginates the MATCHES — the full record set + # is already materialized in Python (the documented design), + # so the filter adds no extra query. + if q: + records = [r for r in records if _record_matches(r, q)] + + total = len(records) + bounded_size = max(1, min(page_size, _MAX_PAGE_SIZE)) + offset = max(0, page) * bounded_size + + _emit_request_shape_attributes( + kind_key, + include=include, + has_since_filter=since is not None, + page_size=bounded_size, + record_count=total, + entity_windows=entity_windows, + path_kind=path_kind, + path_id=path_id, + ) + + return records[offset : offset + bounded_size], total, truncated + + +def activity_endpoint( + api: Any, model_cls: type[Model], uuid_str: str, request_args: Any +) -> Response: + """Body of ``GET /api/v1/{resource}/<uuid>/activity/``. + + Same shape as :func:`superset.versioning.api_helpers.list_versions_endpoint` + for the ``/versions/`` endpoint family. Resolves the path entity, + parses the request query params, runs :func:`get_activity`, and + wraps the result through ``ActivityResponseSchema``. + + *api* is the FAB ``ModelRestApi`` instance (pass ``self`` from the + endpoint method). *request_args* is ``request.args`` from + ``flask.request`` — passed explicitly so the helper is testable + without a live Flask context. + """ + # pylint: disable=import-outside-toplevel + from superset.versioning.schemas import ActivityResponseSchema + + try: + entity, _ = resolve_endpoint_path_entity(api, model_cls, uuid_str) + except PathEntityResponseError as exc: + return exc.response + + try: + params = parse_activity_query_params(request_args) + except ActivityParamsError as exc: + return api.response_400(message=str(exc)) + + records, count, truncated = get_activity( + model_cls, entity.uuid, resolved_entity=entity, **params + ) + payload = ActivityResponseSchema().dump( + {"result": records, "count": count, "truncated": truncated} + ) + return api.response(200, **payload) + + +# ---- Observability ------------------------------------------------------- + +#: Common prefix for every metric this module emits. +_METRIC_PREFIX = "superset.activity_view" + + [email protected] +def _phase_timer(kind_key: str, phase: str) -> Iterator[None]: + """Time the wrapped block and emit + ``superset.activity_view.<kind>.<phase>`` to ``stats_logger_manager``. + Wrapper around :func:`superset.utils.decorators.stats_timing` that + centralises the key construction. + """ + # pylint: disable=import-outside-toplevel + from superset.extensions import stats_logger_manager + from superset.utils.decorators import stats_timing + + with stats_timing( + f"{_METRIC_PREFIX}.{kind_key}.{phase}", + stats_logger_manager.instance, + ): + yield + + +def _emit_request_shape_attributes( + kind_key: str, + *, + include: str, + has_since_filter: bool, + page_size: int, + record_count: int, + entity_windows: list[EntityWindows], + path_kind: str, + path_id: int, +) -> None: + """Emit non-PII shape counters about the request and its result set. + + Emits: include_mode / has_since_filter / page_size / record_count + + per-related-kind entity counts. **No PII**: entity names, diff + content, user identifiers — none of those reach the metric layer. + The counters use ``incr`` (counters) since they're tags, not + latencies; the timing keys above carry the latency dimension. + """ + # pylint: disable=import-outside-toplevel + from superset.extensions import stats_logger_manager + + sl = stats_logger_manager.instance + + # Tag-style metrics: one counter per attribute value. The statsd + # bridge accepts arbitrary strings; downstream dashboards filter by + # the value segment. + sl.incr(f"{_METRIC_PREFIX}.{kind_key}.requests.include_{include}") + sl.incr( + f"{_METRIC_PREFIX}.{kind_key}.requests." + f"has_since_filter_{'true' if has_since_filter else 'false'}" + ) + sl.gauge(f"{_METRIC_PREFIX}.{kind_key}.page_size", float(page_size)) + sl.gauge(f"{_METRIC_PREFIX}.{kind_key}.record_count", float(record_count)) + + # Per-related-kind entity counts. The scope + # list includes the path entity itself (the "self" window); exclude + # it so the gauge reflects only the *related* entities the request + # fanned out to, not "this request touched itself". + by_kind: dict[str, int] = {"Slice": 0, "SqlaTable": 0, "Dashboard": 0} + for api_kind, entity_id, _windows in entity_windows: + if (api_kind, entity_id) == (path_kind, path_id): + continue + if api_kind in by_kind: + by_kind[api_kind] += 1 + sl.gauge( + f"{_METRIC_PREFIX}.{kind_key}.related_entity_count.charts", + float(by_kind["Slice"]), + ) + sl.gauge( + f"{_METRIC_PREFIX}.{kind_key}.related_entity_count.datasets", + float(by_kind["SqlaTable"]), Review Comment: Addressed by subsequent commits; the flagged code is no longer current after the rebase onto master. Resolving as outdated. ########## superset/dashboards/api.py: ########## @@ -2540,6 +2539,85 @@ def get_version(self, uuid_str: str, version_uuid_str: str) -> Response: 404: $ref: '#/components/responses/404' """ - return get_version_endpoint( - self, Dashboard, uuid_str, version_uuid_str, access_kwarg="dashboard" - ) + return get_version_endpoint(self, Dashboard, uuid_str, version_uuid_str) + + @expose("/<uuid_str>/activity/", methods=("GET",)) + @protect() + @safe + @statsd_metrics + @event_logger.log_this_with_context( + action=lambda self, *args, **kwargs: f"{self.__class__.__name__}.activity", + log_to_statsd=False, + ) + def activity(self, uuid_str: str) -> Response: Review Comment: Fixed in 8b19744b85 by mapping the activity endpoint to the standard `get` permission. ########## superset/datasets/api.py: ########## @@ -1808,6 +1807,86 @@ def get_version(self, uuid_str: str, version_uuid_str: str) -> Response: 404: $ref: '#/components/responses/404' """ - return get_version_endpoint( - self, SqlaTable, uuid_str, version_uuid_str, access_kwarg="datasource" - ) + return get_version_endpoint(self, SqlaTable, uuid_str, version_uuid_str) + + @expose("/<uuid_str>/activity/", methods=("GET",)) + @protect() + @safe + @statsd_metrics + @event_logger.log_this_with_context( + action=lambda self, *args, **kwargs: f"{self.__class__.__name__}.activity", + log_to_statsd=False, + ) + def activity(self, uuid_str: str) -> Response: Review Comment: Fixed in 8b19744b85 by mapping the activity endpoint to the standard `get` permission. ########## superset/versioning/activity/queries.py: ########## @@ -0,0 +1,754 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""DB-touching helpers for the activity-view read path. + +All Phase A relationship walks (``charts_attached_to_dashboard``, +``datasets_used_by_chart``, ``batch_datasets_used_by_charts``), +the Phase B change-record fetch (``fetch_change_records`` / +``_select_change_rows_for_kinds``), the name-denormalization helpers +(``_resolve_names_for_kind`` / ``apply_entity_name_denormalization``), the +path-entity resolution helper (``resolve_path_entity``), and the +tombstone-state lookup (``check_entity_tombstones``) live here. + +Each helper is a thin SELECT-and-shape function — no orchestration, +no business logic. Callers in :mod:`scope`, :mod:`render`, and +:mod:`orchestrator` compose them into the end-to-end request. + +**Inline imports.** Continuum's ``version_class`` / ``versioning_manager`` +and the Superset model classes are imported inside each helper because +this package is loaded from ``init_versioning()`` before all SQLAlchemy +mappers are configured. +""" + +from __future__ import annotations + +import logging +from datetime import datetime +from typing import Any +from uuid import UUID + +import sqlalchemy as sa +from flask_appbuilder import Model + +from superset.extensions import db +from superset.versioning.activity.kinds import ( + API_KIND_TO_TABLE, + chunked_ids, + EntityWindows, + load_live_model, + NAME_COLUMN, + NOT_FOUND_EXC, + TABLE_KIND_TO_API, + Window, +) +from superset.versioning.activity.windows import row_within_any_window +from superset.versioning.changes import version_changes_table + +logger = logging.getLogger(__name__) + +# ---- Path-entity resolution ----------------------------------------------- + + +def resolve_path_entity(model_cls: type[Model], entity_uuid: UUID) -> tuple[Any, int]: + """Resolve *entity_uuid* to ``(live_entity, entity_id)`` or raise a + typed 404. + + Soft-delete handling is inherited transparently from + :func:`superset.versioning.queries.find_active_by_uuid` once it + learns to filter out ``deleted_at IS NOT NULL`` rows; at that point + soft-deleted paths will also raise here. + """ + # pylint: disable=import-outside-toplevel + from superset.versioning.queries import find_active_by_uuid + + entity = find_active_by_uuid(model_cls, entity_uuid) + if entity is None: + api_kind = model_cls.__name__ + exc_cls = NOT_FOUND_EXC.get(api_kind) + if exc_cls is None: + raise LookupError( + f"Activity view does not support model class {api_kind!r}" + ) + raise exc_cls(str(entity_uuid)) + return entity, entity.id + + +def first_tracked_tx( + model_cls: type[Model], entity_id: int, entity_uuid: UUID +) -> int | None: + """Return the earliest Continuum transaction_id of the shadow rows + that belong to *this* live entity, matched on ``(id, uuid)``; ``None`` + when the entity has no tracked history yet. + + Used to lower-bound the self-scope window. Matching on the integer + ``id`` alone would inherit a previously hard-deleted entity's history + under id reuse (SQLite/MySQL reuse ``max(id)+1``); pinning the uuid + too — the same discrimination :func:`mark_first_tracked_saves` uses — + scopes the window to the current entity's own transactions. + """ + # pylint: disable=import-outside-toplevel + from sqlalchemy_continuum import version_class + + shadow_tbl = version_class(model_cls).__table__ + stmt = sa.select(sa.func.min(shadow_tbl.c.transaction_id)).where( + shadow_tbl.c.id == entity_id, + shadow_tbl.c.uuid == entity_uuid, + ) + return db.session.connection().execute(stmt).scalar() + + +# ---- Phase A: relationship-traversal queries ------------------------------ + + +def charts_attached_to_dashboard(dashboard_id: int) -> list[tuple[int, Window]]: + """Return ``(slice_id, window)`` for every chart that has ever been on + *dashboard_id*, with each association's validity window in + transaction-id space. + + Reads from ``dashboard_slices_version`` (Continuum's auto-generated + M2M shadow). Rows with ``operation_type = 2`` (DELETE) are excluded + so we don't synthesize a phantom window from a detachment row. + """ + # pylint: disable=import-outside-toplevel + from sqlalchemy_continuum import version_class + + from superset.models.dashboard import Dashboard + + metadata = version_class(Dashboard).__table__.metadata + m2m_tbl = metadata.tables.get("dashboard_slices_version") + if m2m_tbl is None: + return [] + + rows = ( + db.session.connection() + .execute( + sa.select( + m2m_tbl.c.slice_id, + m2m_tbl.c.transaction_id, + m2m_tbl.c.end_transaction_id, + ).where( + m2m_tbl.c.dashboard_id == dashboard_id, + m2m_tbl.c.operation_type != 2, + m2m_tbl.c.slice_id.is_not(None), + ) + ) + .all() + ) + result: list[tuple[int, Window]] = [] + for row in rows: + try: + window = Window(row[1], row[2]) + except ValueError: + # A degenerate shadow row (end_tx <= start_tx) must not 500 the + # endpoint; skip it and leave a breadcrumb for investigation. + logger.warning( + "activity: skipping degenerate dashboard_slices_version row " + "(dashboard_id=%s, slice_id=%s, tx=%s, end_tx=%s)", + dashboard_id, + row[0], + row[1], + row[2], + ) + continue + result.append((row[0], window)) + return result + + +def datasets_used_by_chart(slice_id: int) -> list[tuple[int, Window]]: + """Return ``(datasource_id, window)`` for every dataset that *slice_id* + has ever pointed at, with each association's validity window. + + Single-slice form, used by ``_resolve_chart_scope`` where there + is only one chart to walk. The dashboard-scope path calls + :func:`batch_datasets_used_by_charts` instead so the query fires + once for all slices on the dashboard, not once per slice. + + Reads from ``slices_version`` (the chart parent shadow). Filters to + ``datasource_type = 'table'`` because the activity view only follows + the chart → ``SqlaTable`` dependency edge (not legacy/other + datasources). Rows with ``operation_type = 2`` are excluded. + """ + return batch_datasets_used_by_charts({slice_id}).get(slice_id, []) + + +def batch_datasets_used_by_charts( + slice_ids: set[int], +) -> dict[int, list[tuple[int, Window]]]: + """Batch form of :func:`datasets_used_by_chart`. Returns + ``{slice_id: [(dataset_id, window), ...]}`` in a single query so the + dashboard-scope walker doesn't fire one query per chart on the + dashboard. The previous per-slice shape became O(n_charts) round- + trips, which dominated ``get_activity`` latency on dashboards with + rich history (profile run 2026-05-26 showed `resolve_scope` + accounting for ~1.9s out of 4s p95). + """ + if not slice_ids: + return {} + + # pylint: disable=import-outside-toplevel + from sqlalchemy_continuum import version_class + + from superset.models.slice import Slice + + slices_tbl = version_class(Slice).__table__ + grouped: dict[int, list[tuple[int, Window]]] = {} + # Chunk the IN-clause under SQLite's bind-variable floor (a dashboard can + # carry more charts than the floor allows in one statement). + rows: list[Any] = [] + for chunk in chunked_ids(slice_ids): + rows.extend( + db.session.connection() + .execute( + sa.select( + slices_tbl.c.id, + slices_tbl.c.datasource_id, + slices_tbl.c.transaction_id, + slices_tbl.c.end_transaction_id, + ).where( + slices_tbl.c.id.in_(chunk), + slices_tbl.c.datasource_type == "table", + slices_tbl.c.operation_type != 2, + slices_tbl.c.datasource_id.is_not(None), + ) + ) + .mappings() + .all() + ) + for row in rows: + try: + window = Window(row["transaction_id"], row["end_transaction_id"]) + except ValueError: + # A degenerate shadow row (end_tx <= start_tx) must not 500 the + # endpoint; skip it and leave a breadcrumb for investigation. + logger.warning( + "activity: skipping degenerate slices_version row " + "(slice_id=%s, datasource_id=%s, tx=%s, end_tx=%s)", + row["id"], + row["datasource_id"], + row["transaction_id"], + row["end_transaction_id"], + ) + continue + grouped.setdefault(row["id"], []).append((row["datasource_id"], window)) + return grouped + + +# ---- Phase B: change-record fetch ----------------------------------------- + + +def fetch_change_records( + entity_window_tuples: list[EntityWindows], + since: datetime | None, + until: datetime | None, +) -> tuple[list[dict[str, Any]], bool]: + """Fetch all ``version_changes`` rows matching any of the supplied + entity-window tuples, joined with ``version_transaction`` for + ``issued_at`` and ``user_id``. + + Each tuple is ``(api_kind, entity_id, [(start_tx, end_tx), ...])``; + a record matches when ``entity_kind`` equals the table-stored form + of *api_kind*, ``entity_id`` matches, and ``transaction_id`` falls + inside at least one of the entity's windows. ``since``/``until`` + further restrict by ``issued_at``. + + Implementation: one SELECT per kind with ``entity_id IN (...)`` and + a wide ``transaction_id`` bound (the union of all windows for that + kind). Per-window precision is applied in Python afterward. This + keeps the SQL shape proportional to the number of *kinds* (≤3) and + the bound proportional to the union of windows, not the cross- + product of (entity, window) — which previously generated one OR + clause per (entity, window) pair and hit SQLite's + ``SQLITE_MAX_EXPR_DEPTH`` limit on dashboards with many slices + or many historical attachment windows. + + The visibility filter runs after this function (records + the requester can't read are silently dropped and must not + contribute to ``count``), so the orchestrator paginates in Python + over the filtered list — there is no DB-level page ``OFFSET`` here. + There IS a per-kind safety ceiling (``_MAX_FETCHED_RECORDS``) that + bounds how much history a single request materializes (to at most + ``n_kinds * _MAX_FETCHED_RECORDS``); when a kind exhausts it, the + second return value is ``True``, the returned stream is clamped to a + clean time cut (records older than the highest per-kind fetch floor + are dropped so truncation is a time boundary, not per-kind holes), and + the caller surfaces ``truncated`` on the response. + + Returns ``(records, truncated)``. Records are ordered by + ``(issued_at DESC, transaction_id DESC, sequence DESC)`` — the + secondary keys break ties for the stable-ordering contract. + """ + if not entity_window_tuples: + return [], False + + # Group windows by (table_kind, entity_id) and by table_kind for SQL + # narrowing. The fetch is per-kind; the post-filter is per-entity. + windows_by_entity: dict[tuple[str, int], list[Window]] = {} + ids_by_kind: dict[str, set[int]] = {} + for api_kind, entity_id, windows in entity_window_tuples: + table_kind = API_KIND_TO_TABLE.get(api_kind) + if table_kind is None or not windows: + continue + ids_by_kind.setdefault(table_kind, set()).add(entity_id) + windows_by_entity.setdefault((table_kind, entity_id), []).extend(windows) + + if not ids_by_kind: + return [], False + + # Per-kind transaction_id bounds = the union of that kind's windows. + # Pushing these into the SQL WHERE ensures the per-statement LIMIT + # selects from IN-WINDOW rows. Without it, a related entity whose + # association window is far in the past would have the newest ``limit`` + # (out-of-window) rows fetched and discarded, silently dropping its + # in-window records that lie beyond the limit. ``end_tx = None`` + # (open-ended/current) means no upper bound for that kind. + bounds_by_kind: dict[str, tuple[int, int | None]] = {} + for (table_kind, _entity_id), windows in windows_by_entity.items(): + for w in windows: + cur = bounds_by_kind.get(table_kind) + if cur is None: + bounds_by_kind[table_kind] = (w.start_tx, w.end_tx) + continue + lo = min(cur[0], w.start_tx) + hi = None if (cur[1] is None or w.end_tx is None) else max(cur[1], w.end_tx) + bounds_by_kind[table_kind] = (lo, hi) + + rows, truncated, truncation_floor = _select_change_rows_for_kinds( + ids_by_kind, bounds_by_kind, since, until, _MAX_FETCHED_RECORDS + ) + filtered = [ + row + for row in rows + if row_within_any_window( + row, windows_by_entity.get((row["entity_kind"], row["entity_id"]), []) + ) + ] + if truncated and truncation_floor is not None: + # Collapse truncation to a clean time cut. Below the highest per-kind + # fetch floor at least one kind is missing rows, so surfacing the + # other kinds there would present an incomplete ("nothing else + # changed") picture. Drop everything older than the floor so + # ``truncated=True`` means "complete at/after this instant, nothing + # shown before it" — a time boundary rather than per-kind holes. + filtered = [r for r in filtered if r["issued_at"] >= truncation_floor] Review Comment: Addressed by subsequent commits; the flagged code is no longer current after the rebase onto master. Resolving as outdated. ########## superset/versioning/activity/queries.py: ########## @@ -0,0 +1,754 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""DB-touching helpers for the activity-view read path. + +All Phase A relationship walks (``charts_attached_to_dashboard``, +``datasets_used_by_chart``, ``batch_datasets_used_by_charts``), +the Phase B change-record fetch (``fetch_change_records`` / +``_select_change_rows_for_kinds``), the name-denormalization helpers +(``_resolve_names_for_kind`` / ``apply_entity_name_denormalization``), the +path-entity resolution helper (``resolve_path_entity``), and the +tombstone-state lookup (``check_entity_tombstones``) live here. + +Each helper is a thin SELECT-and-shape function — no orchestration, +no business logic. Callers in :mod:`scope`, :mod:`render`, and +:mod:`orchestrator` compose them into the end-to-end request. + +**Inline imports.** Continuum's ``version_class`` / ``versioning_manager`` +and the Superset model classes are imported inside each helper because +this package is loaded from ``init_versioning()`` before all SQLAlchemy +mappers are configured. +""" + +from __future__ import annotations + +import logging +from datetime import datetime +from typing import Any +from uuid import UUID + +import sqlalchemy as sa +from flask_appbuilder import Model + +from superset.extensions import db +from superset.versioning.activity.kinds import ( + API_KIND_TO_TABLE, + chunked_ids, + EntityWindows, + load_live_model, + NAME_COLUMN, + NOT_FOUND_EXC, + TABLE_KIND_TO_API, + Window, +) +from superset.versioning.activity.windows import row_within_any_window +from superset.versioning.changes import version_changes_table + +logger = logging.getLogger(__name__) + +# ---- Path-entity resolution ----------------------------------------------- + + +def resolve_path_entity(model_cls: type[Model], entity_uuid: UUID) -> tuple[Any, int]: + """Resolve *entity_uuid* to ``(live_entity, entity_id)`` or raise a + typed 404. + + Soft-delete handling is inherited transparently from + :func:`superset.versioning.queries.find_active_by_uuid` once it + learns to filter out ``deleted_at IS NOT NULL`` rows; at that point + soft-deleted paths will also raise here. + """ + # pylint: disable=import-outside-toplevel + from superset.versioning.queries import find_active_by_uuid + + entity = find_active_by_uuid(model_cls, entity_uuid) + if entity is None: + api_kind = model_cls.__name__ + exc_cls = NOT_FOUND_EXC.get(api_kind) + if exc_cls is None: + raise LookupError( + f"Activity view does not support model class {api_kind!r}" + ) + raise exc_cls(str(entity_uuid)) + return entity, entity.id + + +def first_tracked_tx( + model_cls: type[Model], entity_id: int, entity_uuid: UUID +) -> int | None: + """Return the earliest Continuum transaction_id of the shadow rows + that belong to *this* live entity, matched on ``(id, uuid)``; ``None`` + when the entity has no tracked history yet. + + Used to lower-bound the self-scope window. Matching on the integer + ``id`` alone would inherit a previously hard-deleted entity's history + under id reuse (SQLite/MySQL reuse ``max(id)+1``); pinning the uuid + too — the same discrimination :func:`mark_first_tracked_saves` uses — + scopes the window to the current entity's own transactions. + """ + # pylint: disable=import-outside-toplevel + from sqlalchemy_continuum import version_class + + shadow_tbl = version_class(model_cls).__table__ + stmt = sa.select(sa.func.min(shadow_tbl.c.transaction_id)).where( + shadow_tbl.c.id == entity_id, + shadow_tbl.c.uuid == entity_uuid, + ) + return db.session.connection().execute(stmt).scalar() + + +# ---- Phase A: relationship-traversal queries ------------------------------ + + +def charts_attached_to_dashboard(dashboard_id: int) -> list[tuple[int, Window]]: + """Return ``(slice_id, window)`` for every chart that has ever been on + *dashboard_id*, with each association's validity window in + transaction-id space. + + Reads from ``dashboard_slices_version`` (Continuum's auto-generated + M2M shadow). Rows with ``operation_type = 2`` (DELETE) are excluded + so we don't synthesize a phantom window from a detachment row. + """ + # pylint: disable=import-outside-toplevel + from sqlalchemy_continuum import version_class + + from superset.models.dashboard import Dashboard + + metadata = version_class(Dashboard).__table__.metadata + m2m_tbl = metadata.tables.get("dashboard_slices_version") + if m2m_tbl is None: + return [] + + rows = ( + db.session.connection() + .execute( + sa.select( + m2m_tbl.c.slice_id, + m2m_tbl.c.transaction_id, + m2m_tbl.c.end_transaction_id, + ).where( + m2m_tbl.c.dashboard_id == dashboard_id, + m2m_tbl.c.operation_type != 2, + m2m_tbl.c.slice_id.is_not(None), + ) + ) + .all() + ) + result: list[tuple[int, Window]] = [] + for row in rows: + try: + window = Window(row[1], row[2]) + except ValueError: + # A degenerate shadow row (end_tx <= start_tx) must not 500 the + # endpoint; skip it and leave a breadcrumb for investigation. + logger.warning( + "activity: skipping degenerate dashboard_slices_version row " + "(dashboard_id=%s, slice_id=%s, tx=%s, end_tx=%s)", + dashboard_id, + row[0], + row[1], + row[2], + ) + continue + result.append((row[0], window)) + return result + + +def datasets_used_by_chart(slice_id: int) -> list[tuple[int, Window]]: + """Return ``(datasource_id, window)`` for every dataset that *slice_id* + has ever pointed at, with each association's validity window. + + Single-slice form, used by ``_resolve_chart_scope`` where there + is only one chart to walk. The dashboard-scope path calls + :func:`batch_datasets_used_by_charts` instead so the query fires + once for all slices on the dashboard, not once per slice. + + Reads from ``slices_version`` (the chart parent shadow). Filters to + ``datasource_type = 'table'`` because the activity view only follows + the chart → ``SqlaTable`` dependency edge (not legacy/other + datasources). Rows with ``operation_type = 2`` are excluded. + """ + return batch_datasets_used_by_charts({slice_id}).get(slice_id, []) + + +def batch_datasets_used_by_charts( + slice_ids: set[int], +) -> dict[int, list[tuple[int, Window]]]: + """Batch form of :func:`datasets_used_by_chart`. Returns + ``{slice_id: [(dataset_id, window), ...]}`` in a single query so the + dashboard-scope walker doesn't fire one query per chart on the + dashboard. The previous per-slice shape became O(n_charts) round- + trips, which dominated ``get_activity`` latency on dashboards with + rich history (profile run 2026-05-26 showed `resolve_scope` + accounting for ~1.9s out of 4s p95). + """ + if not slice_ids: + return {} + + # pylint: disable=import-outside-toplevel + from sqlalchemy_continuum import version_class + + from superset.models.slice import Slice + + slices_tbl = version_class(Slice).__table__ + grouped: dict[int, list[tuple[int, Window]]] = {} + # Chunk the IN-clause under SQLite's bind-variable floor (a dashboard can + # carry more charts than the floor allows in one statement). + rows: list[Any] = [] + for chunk in chunked_ids(slice_ids): + rows.extend( + db.session.connection() + .execute( + sa.select( + slices_tbl.c.id, + slices_tbl.c.datasource_id, + slices_tbl.c.transaction_id, + slices_tbl.c.end_transaction_id, + ).where( + slices_tbl.c.id.in_(chunk), + slices_tbl.c.datasource_type == "table", + slices_tbl.c.operation_type != 2, + slices_tbl.c.datasource_id.is_not(None), + ) + ) + .mappings() + .all() + ) + for row in rows: + try: + window = Window(row["transaction_id"], row["end_transaction_id"]) + except ValueError: + # A degenerate shadow row (end_tx <= start_tx) must not 500 the + # endpoint; skip it and leave a breadcrumb for investigation. + logger.warning( + "activity: skipping degenerate slices_version row " + "(slice_id=%s, datasource_id=%s, tx=%s, end_tx=%s)", + row["id"], + row["datasource_id"], + row["transaction_id"], + row["end_transaction_id"], + ) + continue + grouped.setdefault(row["id"], []).append((row["datasource_id"], window)) + return grouped + + +# ---- Phase B: change-record fetch ----------------------------------------- + + +def fetch_change_records( + entity_window_tuples: list[EntityWindows], + since: datetime | None, + until: datetime | None, +) -> tuple[list[dict[str, Any]], bool]: + """Fetch all ``version_changes`` rows matching any of the supplied + entity-window tuples, joined with ``version_transaction`` for + ``issued_at`` and ``user_id``. + + Each tuple is ``(api_kind, entity_id, [(start_tx, end_tx), ...])``; + a record matches when ``entity_kind`` equals the table-stored form + of *api_kind*, ``entity_id`` matches, and ``transaction_id`` falls + inside at least one of the entity's windows. ``since``/``until`` + further restrict by ``issued_at``. + + Implementation: one SELECT per kind with ``entity_id IN (...)`` and + a wide ``transaction_id`` bound (the union of all windows for that + kind). Per-window precision is applied in Python afterward. This + keeps the SQL shape proportional to the number of *kinds* (≤3) and + the bound proportional to the union of windows, not the cross- + product of (entity, window) — which previously generated one OR + clause per (entity, window) pair and hit SQLite's + ``SQLITE_MAX_EXPR_DEPTH`` limit on dashboards with many slices + or many historical attachment windows. + + The visibility filter runs after this function (records + the requester can't read are silently dropped and must not + contribute to ``count``), so the orchestrator paginates in Python + over the filtered list — there is no DB-level page ``OFFSET`` here. + There IS a per-kind safety ceiling (``_MAX_FETCHED_RECORDS``) that + bounds how much history a single request materializes (to at most + ``n_kinds * _MAX_FETCHED_RECORDS``); when a kind exhausts it, the + second return value is ``True``, the returned stream is clamped to a + clean time cut (records older than the highest per-kind fetch floor + are dropped so truncation is a time boundary, not per-kind holes), and + the caller surfaces ``truncated`` on the response. + + Returns ``(records, truncated)``. Records are ordered by + ``(issued_at DESC, transaction_id DESC, sequence DESC)`` — the + secondary keys break ties for the stable-ordering contract. + """ + if not entity_window_tuples: + return [], False + + # Group windows by (table_kind, entity_id) and by table_kind for SQL + # narrowing. The fetch is per-kind; the post-filter is per-entity. + windows_by_entity: dict[tuple[str, int], list[Window]] = {} + ids_by_kind: dict[str, set[int]] = {} + for api_kind, entity_id, windows in entity_window_tuples: + table_kind = API_KIND_TO_TABLE.get(api_kind) + if table_kind is None or not windows: + continue + ids_by_kind.setdefault(table_kind, set()).add(entity_id) + windows_by_entity.setdefault((table_kind, entity_id), []).extend(windows) + + if not ids_by_kind: + return [], False + + # Per-kind transaction_id bounds = the union of that kind's windows. + # Pushing these into the SQL WHERE ensures the per-statement LIMIT + # selects from IN-WINDOW rows. Without it, a related entity whose + # association window is far in the past would have the newest ``limit`` + # (out-of-window) rows fetched and discarded, silently dropping its + # in-window records that lie beyond the limit. ``end_tx = None`` + # (open-ended/current) means no upper bound for that kind. + bounds_by_kind: dict[str, tuple[int, int | None]] = {} + for (table_kind, _entity_id), windows in windows_by_entity.items(): + for w in windows: + cur = bounds_by_kind.get(table_kind) + if cur is None: + bounds_by_kind[table_kind] = (w.start_tx, w.end_tx) + continue + lo = min(cur[0], w.start_tx) + hi = None if (cur[1] is None or w.end_tx is None) else max(cur[1], w.end_tx) + bounds_by_kind[table_kind] = (lo, hi) + + rows, truncated, truncation_floor = _select_change_rows_for_kinds( + ids_by_kind, bounds_by_kind, since, until, _MAX_FETCHED_RECORDS + ) + filtered = [ + row + for row in rows + if row_within_any_window( + row, windows_by_entity.get((row["entity_kind"], row["entity_id"]), []) + ) + ] + if truncated and truncation_floor is not None: + # Collapse truncation to a clean time cut. Below the highest per-kind + # fetch floor at least one kind is missing rows, so surfacing the + # other kinds there would present an incomplete ("nothing else + # changed") picture. Drop everything older than the floor so + # ``truncated=True`` means "complete at/after this instant, nothing + # shown before it" — a time boundary rather than per-kind holes. + filtered = [r for r in filtered if r["issued_at"] >= truncation_floor] + # Sort key must be TOTAL so pagination is stable across requests: two + # records from different entities can share (issued_at, transaction_id, + # sequence), so append (entity_kind, entity_id) to break remaining ties + # deterministically. Without these the relative order of tied records + # depends on set-iteration order and a record could shift pages. + filtered.sort( + key=lambda r: ( + r["issued_at"], + r["transaction_id"], + r["sequence"], + r["entity_kind"], + r["entity_id"], + ), + reverse=True, + ) + return filtered, truncated + + +def _select_change_rows_for_kinds( + ids_by_kind: dict[str, set[int]], + bounds_by_kind: dict[str, tuple[int, int | None]], + since: datetime | None, + until: datetime | None, + limit: int, +) -> tuple[list[dict[str, Any]], bool, datetime | None]: + """Fire one SELECT per entity_kind with ``entity_id IN (...)``; + concatenate the results. Each SELECT joins ``version_transaction`` + + ``ab_user`` so the orchestrator has the columns it needs for + decoration. + + Per-kind, not one query: SQLAlchemy's ``tuple_(entity_kind, + entity_id).in_(...)`` would collapse the three queries into one, + but its SQL emission is not portable across Postgres, MySQL, and + SQLite. The per-kind shape is the correct trade-off given + Superset's multi-dialect requirement (at most 3 round-trips per + request, bounded by the kind taxonomy). Do not "optimise" into a + composite-tuple IN clause without verifying the SQL on all three + dialects. + + **Init-order dependency.** ``tx_tbl.c.action_kind`` resolves only + after ``init_versioning()`` has run — the column is appended onto + Continuum's transaction Table by + ``superset.versioning.factory.VersionTransactionFactory`` at app + start via ``append_column`` + ``add_property``. This helper is + safe to call from request-path code because the app is fully + initialised by then; calling it from a script that imports the + versioning package without going through ``init_versioning()`` + will raise ``AttributeError`` on the ``action_kind`` attribute + access below.""" + # pylint: disable=import-outside-toplevel + from sqlalchemy_continuum import versioning_manager + + from superset import security_manager + + tx_tbl = versioning_manager.transaction_cls.__table__ + user_tbl = security_manager.user_model.__table__ + vc = version_changes_table + join_tree = vc.join(tx_tbl, vc.c.transaction_id == tx_tbl.c.id).outerjoin( + user_tbl, tx_tbl.c.user_id == user_tbl.c.id + ) + select_cols = ( + vc.c.transaction_id, + vc.c.entity_kind, + vc.c.entity_id, + vc.c.sequence, + vc.c.kind, + vc.c.operation, + vc.c.path, + vc.c.from_value, + vc.c.to_value, + tx_tbl.c.issued_at, + tx_tbl.c.user_id, + # ``action_kind`` is the high-level avenue (restore / import / + # clone / NULL=ordinary save) stamped by the originating + # command via the change-record listener. All records sharing a + # ``transaction_id`` share the same value. The column is + # declared on the Continuum Table by ``VersionTransactionFactory``, + # so ``tx_tbl.c.action_kind`` resolves cleanly here. See + # the three change-record dimensions. + tx_tbl.c.action_kind, + user_tbl.c.id.label("changed_by_id"), + user_tbl.c.first_name, + user_tbl.c.last_name, + ) + + out: list[dict[str, Any]] = [] + truncated = False + truncation_floors: list[datetime] = [] + for table_kind, entity_ids in ids_by_kind.items(): + # The fetch budget is per KIND, not per id-chunk. A kind whose + # entity_ids span several bind-variable chunks shares one ``limit`` + # across those chunks, so the rows a single request materializes are + # bounded by ``n_kinds * limit`` (<= 3 * _MAX_FETCHED_RECORDS), not + # ``n_chunks * limit``. ``entity_ids`` is chunked to stay inside + # SQLite's ``SQLITE_MAX_VARIABLE_NUMBER`` floor (default 999 in many + # builds); Postgres + MySQL accept the full list but the chunk is + # dialect-agnostic for simplicity. + kind_rows: list[dict[str, Any]] = [] + for chunk in chunked_ids(entity_ids): + remaining = limit - len(kind_rows) + if remaining <= 0: + break + stmt = ( + sa.select(*select_cols) + .select_from(join_tree) + .where( + vc.c.entity_kind == table_kind, + vc.c.entity_id.in_(chunk), + ) + ) + # Bound by the kind's window union so the LIMIT picks in-window + # rows (see fetch_change_records). The per-entity window filter + # still runs in Python afterwards for exact membership. + tx_lo, tx_hi = bounds_by_kind[table_kind] + stmt = stmt.where(vc.c.transaction_id >= tx_lo) + if tx_hi is not None: + stmt = stmt.where(vc.c.transaction_id < tx_hi) + if since is not None: + stmt = stmt.where(tx_tbl.c.issued_at >= since) + if until is not None: + stmt = stmt.where(tx_tbl.c.issued_at < until) + # Match fetch_change_records' final Python sort key order + # (issued_at, transaction_id, sequence, entity_id) so the + # per-kind budget keeps exactly the rows the final sort ranks + # highest. entity_kind is constant within a per-kind statement. + stmt = stmt.order_by( + tx_tbl.c.issued_at.desc(), + vc.c.transaction_id.desc(), + vc.c.sequence.desc(), + vc.c.entity_id.desc(), + ).limit(remaining) + kind_rows.extend( + dict(row) + for row in db.session.connection().execute(stmt).mappings().all() + ) Review Comment: Addressed by subsequent commits; the flagged code is no longer current after the rebase onto master. Resolving as outdated. -- 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]
