(superset) branch master updated: fix(versioning): scope stream_results to the activity SELECTs instead of mutating the session connection (#44261)

[ri...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

richardfogaca pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/superset.git


The following commit(s) were added to refs/heads/master by this push:
     new c426186a2e6 fix(versioning): scope stream_results to the activity SELECTs instead of mutating the session connection (#44261)
c426186a2e6 is described below

commit c426186a2e664301dfa64400bc4a9ac1ec35b784
Author: Mike Bridge <mi...@preset.io>
AuthorDate: Thu Sep 17 12:27:47 2026 -0600

    fix(versioning): scope stream_results to the activity SELECTs instead of mutating the session connection (#44261)
    
    Co-authored-by: Claude Fable 5 <no...@anthropic.com>
---
 superset/versioning/activity/queries.py            |  11 +-
 .../versioning/activity_write_after_read_tests.py  | 124 +++++++++++++++++++++
 2 files changed, 133 insertions(+), 2 deletions(-)

diff --git a/superset/versioning/activity/queries.py b/superset/versioning/activity/queries.py
index 5c6476b65e1..e29c242baff 100644
--- a/superset/versioning/activity/queries.py
+++ b/superset/versioning/activity/queries.py
@@ -418,10 +418,17 @@ def _select_change_rows_for_kinds(
                 vc.c.sequence.desc(),
                 vc.c.entity_id.desc(),
             )
+            # stream_results rides the STATEMENT, never the connection:
+            # ``Connection.execution_options`` mutates the session's
+            # connection in place, permanently flipping every later
+            # statement on the request into a server-side cursor — on
+            # PostgreSQL that wraps subsequent INSERTs/SAVEPOINTs in
+            # ``DECLARE ... CURSOR FOR`` and they fail with a syntax
+            # error (observed: Continuum's transaction insert and the
+            # DBEventLogger write after an activity read; sc-120955).
             result = (
                 db.session.connection()
-                .execution_options(stream_results=True)
-                .execute(stmt)
+                .execute(stmt.execution_options(stream_results=True))
                 .mappings()
             )
             ordinal = _merge_result_into_heap(
diff --git a/tests/integration_tests/versioning/activity_write_after_read_tests.py b/tests/integration_tests/versioning/activity_write_after_read_tests.py
new file mode 100644
index 00000000000..8ef1790dee4
--- /dev/null
+++ b/tests/integration_tests/versioning/activity_write_after_read_tests.py
@@ -0,0 +1,124 @@
+# 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.
+"""sc-120955: an activity read must not poison the session for later
+writes. ``fetch_change_records`` used to flip the session connection
+into server-side-cursor mode in place; on PostgreSQL every subsequent
+INSERT/SAVEPOINT on the request then failed with a syntax error
+(``DECLARE ... CURSOR FOR INSERT``) — the DBEventLogger write after any
+activity read being the production shape."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from flask_appbuilder.security.sqla.models import User
+from sqlalchemy import event
+from sqlalchemy.engine import Connection, ExecutionContext
+
+from superset.extensions import db
+from superset.models.slice import Slice
+from superset.utils.core import override_user
+from superset.versioning.activity.orchestrator import get_activity
+from tests.integration_tests.base_tests import SupersetTestCase
+
+
+def _admin_user() -> User:
+    # pylint: disable=import-outside-toplevel
+    from superset import security_manager
+
+    return (
+        db.session.query(security_manager.user_model)
+        .filter_by(username="admin")
+        .one_or_none()
+    ) or security_manager.add_user(
+        "admin",
+        "admin",
+        "user",
+        "admin@fab.org",
+        security_manager.find_role("Admin"),
+        password="general",  # noqa: S106 — test-only fixture credential
+    )
+
+
+class TestActivityReadThenWrite(SupersetTestCase):
+    def test_write_on_the_same_session_after_an_activity_read(self) -> None:
+        """The DBEventLogger shape: read activity, then INSERT + commit on
+        the SAME session. With the connection mutated into stream_results
+        mode this failed on PostgreSQL before any assertion ran."""
+        slc: Slice = Slice(
+            slice_name="sc120955_read_then_write",
+            datasource_type="table",
+            datasource_id=1,
+            params="{}",
+        )
+        db.session.add(slc)
+        db.session.commit()
+        try:
+            # The read that used to poison the connection. The entity has
+            # history (its own INSERT), so the streaming fetch executes.
+            slc.slice_name = "sc120955_read_then_write_edited"
+            db.session.commit()
+            streamed_cursors: list[str | None] = []
+
+            def observe_change_select(
+                connection: Connection,
+                cursor: Any,
+                statement: str,
+                parameters: Any,
+                context: ExecutionContext,
+                executemany: bool,
+            ) -> None:
+                """Observe the change SELECT even if streaming is removed."""
+                if statement.lstrip().upper().startswith("SELECT") and (
+                    "version_changes" in statement
+                ):
+                    streamed_cursors.append(getattr(cursor, "name", None))
+
+            records: list[dict[str, Any]]
+            event.listen(db.engine, "before_cursor_execute", observe_change_select)
+            try:
+                with override_user(_admin_user()):
+                    records, _, _ = get_activity(Slice, slc.uuid, resolved_entity=slc)
+            finally:
+                event.remove(db.engine, "before_cursor_execute", observe_change_select)
+            assert records is not None
+            assert streamed_cursors, "The activity read must execute its change SELECT"
+            if db.engine.dialect.name == "postgresql":
+                assert all(streamed_cursors), "Activity SELECTs must use named cursors"
+            assert (
+                not db.session.connection()
+                .get_execution_options()
+                .get("stream_results", False)
+            )
+
+            # A write through the same session must still work — this is
+            # what Continuum's transaction INSERT, SAVEPOINTs, and the
+            # DBEventLogger all do after the endpoint's read.
+            follower: Slice = Slice(
+                slice_name="sc120955_follow_up_write",
+                datasource_type="table",
+                datasource_id=1,
+                params="{}",
+            )
+            db.session.add(follower)
+            db.session.commit()
+            assert follower.id is not None
+            db.session.delete(follower)
+            db.session.commit()
+        finally:
+            db.session.delete(slc)
+            db.session.commit()