Expose query stats through a cursor callback#623
Conversation
Add an optional stats_callback to Connection.cursor() that is invoked with a copy of the query stats every time the client processes a response from the coordinator: once right after the query is submitted (so the query id and initial state are available while the query is still running) and again on every subsequent poll. This mirrors the progress monitor exposed by the Trino JDBC driver and lets callers log or monitor a query instead of only being able to inspect it after all rows have been fetched.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
This comment was marked as outdated.
|
@coderabbitai full review |
✅ Action performedFull review finished. |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@trino/client.py`:
- Around line 1026-1030: Update _report_stats to pass a deep copy of self._stats
to _stats_callback, preventing mutations to nested dictionaries and lists from
affecting internal query state. Add the copy module import if needed, while
preserving the existing callback guard.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 5dbd763b-ae94-4537-84e3-823a5b1795f3
📒 Files selected for processing (4)
README.mdtests/integration/test_dbapi_integration.pytrino/client.pytrino/dbapi.py
| def _report_stats(self) -> None: | ||
| if self._stats_callback is not None: | ||
| # Pass a copy so the callback cannot mutate internal query state. | ||
| self._stats_callback(dict(self._stats)) | ||
|
|
There was a problem hiding this comment.
🎯 Functional Correctness | 🟡 Minor | ⚡ Quick win
Use a deep copy to prevent mutation of nested query state.
The dictionary self._stats contains deeply nested JSON structures (such as rootStage and operatorSummaries) returned by the Trino coordinator. A shallow dict() copy only protects the top-level keys; if the callback mutates a nested dictionary or list, it will modify the client's internal query state.
To fully enforce the isolation contract promised in the documentation and comments, use a deep copy.
🛡️ Proposed fix to use deepcopy
Make sure to add import copy at the top of the file if it is not already imported.
def _report_stats(self) -> None:
if self._stats_callback is not None:
- # Pass a copy so the callback cannot mutate internal query state.
- self._stats_callback(dict(self._stats))
+ # Pass a deep copy so the callback cannot mutate internal query state.
+ import copy
+ self._stats_callback(copy.deepcopy(self._stats))📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| def _report_stats(self) -> None: | |
| if self._stats_callback is not None: | |
| # Pass a copy so the callback cannot mutate internal query state. | |
| self._stats_callback(dict(self._stats)) | |
| def _report_stats(self) -> None: | |
| if self._stats_callback is not None: | |
| # Pass a deep copy so the callback cannot mutate internal query state. | |
| import copy | |
| self._stats_callback(copy.deepcopy(self._stats)) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@trino/client.py` around lines 1026 - 1030, Update _report_stats to pass a
deep copy of self._stats to _stats_callback, preventing mutations to nested
dictionaries and lists from affecting internal query state. Add the copy module
import if needed, while preserving the existing callback guard.
There was a problem hiding this comment.
Seems like a reasonable suggestion
There was a problem hiding this comment.
@azawlocki-sbdt, agreed. Since copy is already imported at module scope, the minimal change is to call copy.deepcopy(self._stats) in _report_stats(), preventing callback mutations of nested statistics from affecting the query’s internal state.
There was a problem hiding this comment.
@hashhar, can you also add a test that checks that the callback cannot modify the original stats?
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
fd -t f -e py . tests trino | sort
printf '\n--- TrinoQuery tests and callback coverage ---\n'
rg -n -C 4 --glob '*.py' \
'(class\s+Test.*TrinoQuery|def\s+test_.*(?:TrinoQuery|stats_callback)|TrinoQuery\s*\()' \
tests trino
printf '\n--- Existing mock response/status helpers ---\n'
rg -n -C 3 --glob '*.py' \
'(Mock|mock|MagicMock|TrinoStatus\s*\(|status\.stats|_update_state\s*\()' \
testsLength of output: 49444
🏁 Script executed:
#!/bin/bash
set -euo pipefail
printf '%s\n' '--- tests/unit/test_client.py imports ---'
sed -n '1,100p' tests/unit/test_client.py
printf '%s\n' '--- Existing TrinoQuery test block ---'
sed -n '1060,1160p' tests/unit/test_client.pyLength of output: 6894
@azawlocki-sbdt, yes. The regression test will mutate nested rootStage/operatorSummaries data in the callback and assert the query’s internal stats are unaffected. It will fail with the current shallow copy and pass with deepcopy.
- Use
copy.deepcopy()when reporting callback statistics. - Add a focused unit regression test in
tests/unit/test_client.py. - Validate the targeted test and Python formatting/type checks.
🧬 Code Graph Analysis Results
Relevant code snippets
trino/client.py (around lines 410-470) — TrinoQuery captures the callback
class TrinoQuery:
"""Represent the execution of a SQL statement by Trino."""
def __init__(
self,
request: TrinoRequest,
query: str,
legacy_primitive_types: bool = False,
fetch_mode: Literal["mapped", "segments"] = "mapped",
stats_callback: Optional[Callable[[Dict[str, Any]], None]] = None
) -> None:
self._query_id: Optional[str] = None
self._stats: Dict[Any, Any] = {}
...
self._stats_callback = stats_callbacktrino/client.py (around lines 500-535) — _update_state mutates internal stats, which the callback sees
def _update_state(self, status):
self._stats.update(status.stats)
self._update_type = status.update_type
self._update_count = status.update_count
self._next_uri = status.next_uri
if not self._row_mapper and status.columns:
self._row_mapper = RowMapperFactory().create(columns=status.columns,
legacy_primitive_types=self._legacy_primitive_types)
if status.columns:
self._columns = status.columnstrino/client.py (around lines 536-545) — _report_stats passes a copy so the callback can’t mutate _stats
def _report_stats(self) -> None:
if self._stats_callback is not None:
# Pass a copy so the callback cannot mutate internal query state.
self._stats_callback(dict(self._stats))trino/client.py (around lines 445-505) — execute() calls _report_stats() after updating state
status = self._request.process(response)
...
self._stats.update({"queryId": self.query_id})
self._update_state(status)
...
self._report_stats()trino/client.py (around lines 560-620) — fetch() also calls _report_stats() after updating state
status = self._request.process(response)
self._update_state(status)
self._report_stats()
if status.next_uri is None:
self._finished = Truetrino/client.py (around lines 475-485) — external reads of internal stats (what the test would assert)
`@property`
def stats(self):
return self._statstrino/dbapi.py (provided line numbers: 633-666) — DB-API wires stats_callback into TrinoQuery
self._query = trino.client.TrinoQuery(
self._request,
query=operation,
legacy_primitive_types=self._legacy_primitive_types,
stats_callback=self._stats_callback
)
self._iterator = iter(self._query.execute())- 📌 Create a pull request with these changes
| raise trino.exceptions.TrinoConnectionError("failed to fetch: {}".format(e)) | ||
| status = self._request.process(response) | ||
| self._update_state(status) | ||
| self._report_stats() |
There was a problem hiding this comment.
self._report_stats() always follows self._update_state(status), I'd consider calling self._report_stats() from inside self._update_status(status).
Feel free to resolve this comment if you think it's better to leave the calls outside.
2 participants
Description
Add an optional stats_callback to Connection.cursor() that is invoked with a copy of the query stats every time the client processes a response from the coordinator: once right after the query is submitted (so the query id and initial state are available while the query is still running) and again on every subsequent poll.
This mirrors the progress monitor exposed by the Trino JDBC driver and lets callers log or monitor a query instead of only being able to inspect it after all rows have been fetched.
Non-technical explanation
Expose query stats through a cursor callback. Fixes #578 .
Release notes
( ) This is not user-visible or docs only and no release notes are required.
(x) Release notes are required, please propose a release note for me.
( ) Release notes are required, with the following suggested text: