Fix record/replay support for cursor.stats in Snowflake adapter (#1572)

Author: emmyoop

dbt-snowflake/.changes/unreleased/Fixes-20260123-153253.yaml

@@ -0,0 +1,6 @@
+kind: Fixes
+body: Fix record/replay support for cursor.stats attribute access.
+time: 2026-01-23T15:32:53.925706-05:00
+custom:
+  Author: emmyoop
+  Issue: "1573"

dbt-snowflake/src/dbt/adapters/snowflake/record/cursor/cursor.py

@@ -3,11 +3,12 @@
 from dbt.adapters.record import RecordReplayCursor
 from dbt.adapters.snowflake.record.cursor.sfqid import CursorGetSfqidRecord
 from dbt.adapters.snowflake.record.cursor.sqlstate import CursorGetSqlStateRecord
+from dbt.adapters.snowflake.record.cursor.stats import CursorGetStatsRecord
 
 
 class SnowflakeRecordReplayCursor(RecordReplayCursor):
-    """A custom extension of RecordReplayCursor that adds the sqlstate
-    and sfqid properties which are specific to snowflake-connector."""
+    """A custom extension of RecordReplayCursor that adds the sqlstate,
+    sfqid, and stats properties which are specific to snowflake-connector."""
 
     @property
     @record_function(CursorGetSqlStateRecord, method=True, id_field_name="connection_name")
@@ -18,3 +19,8 @@ def sqlstate(self):
     @record_function(CursorGetSfqidRecord, method=True, id_field_name="connection_name")
     def sfqid(self):
         return self.native_cursor.sfqid
+
+    @property
+    @record_function(CursorGetStatsRecord, method=True, id_field_name="connection_name")
+    def stats(self):
+        return self.native_cursor.stats

dbt-snowflake/src/dbt/adapters/snowflake/record/cursor/stats.py

@@ -0,0 +1,72 @@
+import dataclasses
+from typing import Any, Optional
+
+from dbt_common.record import Record, Recorder
+
+
+@dataclasses.dataclass
+class CursorGetStatsParams:
+    connection_name: str
+
+
+@dataclasses.dataclass
+class CursorGetStatsResult:
+    """Captures the stats from SnowflakeCursor.stats (QueryResultStats).
+
+    These stats are available in snowflake-connector-python >= 4.2.0 and provide
+    detailed DML operation information.
+    """
+
+    num_rows_inserted: Optional[int]
+    num_rows_deleted: Optional[int]
+    num_rows_updated: Optional[int]
+    num_dml_duplicates: Optional[int]
+
+
+class StatsProxy:
+    """A proxy object that mimics the QueryResultStats interface for replay mode."""
+
+    def __init__(self, result: Optional[CursorGetStatsResult]) -> None:
+        self._result = result
+
+    @property
+    def num_rows_inserted(self) -> Optional[int]:
+        return self._result.num_rows_inserted if self._result else None
+
+    @property
+    def num_rows_deleted(self) -> Optional[int]:
+        return self._result.num_rows_deleted if self._result else None
+
+    @property
+    def num_rows_updated(self) -> Optional[int]:
+        return self._result.num_rows_updated if self._result else None
+
+    @property
+    def num_dml_duplicates(self) -> Optional[int]:
+        return self._result.num_dml_duplicates if self._result else None
+
+
+@Recorder.register_record_type
+class CursorGetStatsRecord(Record):
+    params_cls = CursorGetStatsParams
+    result_cls = CursorGetStatsResult
+    group = "Database"
+
+    @classmethod
+    def _record_result(cls, result: Any) -> Optional[CursorGetStatsResult]:
+        """Convert the native stats object to our result dataclass."""
+        if result is None:
+            return None
+        return CursorGetStatsResult(
+            num_rows_inserted=getattr(result, "num_rows_inserted", None),
+            num_rows_deleted=getattr(result, "num_rows_deleted", None),
+            num_rows_updated=getattr(result, "num_rows_updated", None),
+            num_dml_duplicates=getattr(result, "num_dml_duplicates", None),
+        )
+
+    @classmethod
+    def _replay_result(cls, result: Optional[CursorGetStatsResult]) -> Optional[StatsProxy]:
+        """Convert our result dataclass back to a stats-like object for replay."""
+        if result is None:
+            return None
+        return StatsProxy(result)

dbt-snowflake/tests/unit/test_record_cursor.py

@@ -0,0 +1,150 @@
+from dbt.adapters.snowflake.connections import SnowflakeConnectionManager
+from dbt.adapters.snowflake.record.cursor.cursor import SnowflakeRecordReplayCursor
+
+
+class MockStats:
+    """Mock object that mimics snowflake-connector-python's QueryResultStats."""
+
+    def __init__(
+        self,
+        num_rows_inserted=100,
+        num_rows_deleted=10,
+        num_rows_updated=5,
+        num_dml_duplicates=2,
+    ):
+        self.num_rows_inserted = num_rows_inserted
+        self.num_rows_deleted = num_rows_deleted
+        self.num_rows_updated = num_rows_updated
+        self.num_dml_duplicates = num_dml_duplicates
+
+
+class MockCursor:
+    """Mock cursor that mimics snowflake-connector-python's SnowflakeCursor."""
+
+    def __init__(self, stats=None):
+        self._stats = stats
+
+    @property
+    def rowcount(self) -> int:
+        return 42
+
+    @property
+    def sqlstate(self) -> str:
+        return "00000"
+
+    @property
+    def sfqid(self) -> str:
+        return "01abc123-0001-abcd-0000-00012345abcd"
+
+    @property
+    def stats(self):
+        return self._stats
+
+    def execute(self, operation, parameters=None) -> None:
+        pass
+
+    @property
+    def unexpected_prop(self) -> bool:
+        return True
+
+    def unexpected_func(self) -> int:
+        return 1
+
+
+class MockConnection:
+    name = "test_connection"
+
+
+def test_snowflake_record_cursor_sqlstate():
+    """Test that the sqlstate property works correctly."""
+    recorded_cursor = SnowflakeRecordReplayCursor(MockCursor(), MockConnection())  # type: ignore
+    assert recorded_cursor.sqlstate == "00000"
+
+
+def test_snowflake_record_cursor_sfqid():
+    """Test that the sfqid property works correctly."""
+    recorded_cursor = SnowflakeRecordReplayCursor(MockCursor(), MockConnection())  # type: ignore
+    assert recorded_cursor.sfqid == "01abc123-0001-abcd-0000-00012345abcd"
+
+
+def test_snowflake_record_cursor_stats():
+    """Test that the stats property works correctly."""
+    mock_stats = MockStats()
+    recorded_cursor = SnowflakeRecordReplayCursor(
+        MockCursor(stats=mock_stats), MockConnection()
+    )  # type: ignore
+
+    stats = recorded_cursor.stats
+    assert stats.num_rows_inserted == 100
+    assert stats.num_rows_deleted == 10
+    assert stats.num_rows_updated == 5
+    assert stats.num_dml_duplicates == 2
+
+
+def test_snowflake_record_cursor_stats_none():
+    """Test that the stats property handles None correctly."""
+    recorded_cursor = SnowflakeRecordReplayCursor(
+        MockCursor(stats=None), MockConnection()
+    )  # type: ignore
+
+    assert recorded_cursor.stats is None
+
+
+def test_snowflake_record_cursor_inherited_properties():
+    """Test that inherited properties from RecordReplayCursor work correctly."""
+    recorded_cursor = SnowflakeRecordReplayCursor(MockCursor(), MockConnection())  # type: ignore
+
+    # Test inherited rowcount property
+    assert recorded_cursor.rowcount == 42
+
+    # Test inherited execute method
+    recorded_cursor.execute("SELECT 1")
+
+
+def test_snowflake_record_cursor_unexpected_access():
+    """Test that unexpected property/method access fires a warning but still works."""
+    recorded_cursor = SnowflakeRecordReplayCursor(MockCursor(), MockConnection())  # type: ignore
+
+    events = []
+    # Mock event firing
+    recorded_cursor._fire_event = events.append
+
+    # Test that an unexpected property works, but fires a warning
+    assert recorded_cursor.unexpected_prop is True
+    assert len(events) == 1
+    assert events[0].__class__.__name__ == "RecordReplayIssue"
+    assert "unexpected_prop" in events[0].msg
+    events.clear()
+
+    # Test that an unexpected function works, but fires a warning
+    assert recorded_cursor.unexpected_func() == 1
+    assert len(events) == 1
+    assert events[0].__class__.__name__ == "RecordReplayIssue"
+    assert "unexpected_func" in events[0].msg
+
+
+def test_get_response_no_unexpected_access_warnings():
+    """Ensure get_response() doesn't trigger any unexpected attribute access warnings.
+
+    This is a regression test. If new cursor attributes are accessed in get_response()
+    without being added to SnowflakeRecordReplayCursor, this test will fail.
+    """
+    events = []
+
+    # Test with stats present
+    mock_cursor = MockCursor(stats=MockStats())
+    recorded_cursor = SnowflakeRecordReplayCursor(mock_cursor, MockConnection())  # type: ignore
+    recorded_cursor._fire_event = events.append
+
+    # Call get_response - this is the actual code path
+    response = SnowflakeConnectionManager.get_response(recorded_cursor)
+
+    # Verify no unexpected access warnings were fired
+    assert len(events) == 0, (
+        f"Unexpected attribute access in get_response(): {[e.msg for e in events]}. "
+        "Add the missing attribute(s) to SnowflakeRecordReplayCursor."
+    )
+
+    # Verify the response was created successfully
+    assert response is not None
+    assert response.code == "00000"  # SQL success state from mock cursor