Send heartbeats while the caller consumes fetched rows - #632
Conversation
2cd7679 to
904d56e
Compare
hashhar
changed the title
46d44b8 to
fb4e162
Compare
|
@hashhar are heartbeats sent out of band or they are blocking the main thread/execution? |
| # requests.Session right after this returns, and that session is not thread safe. | ||
| self._stop_event.set() | ||
| if self._thread is not None: | ||
| self._thread.join() |
There was a problem hiding this comment.
I don't know if it's serious but the join() may block if the joined thread happens to do the heartbeat HEAD request. This shouldn't take long normally, but I think there's a retry mechanism with exponential backoff so if the server is not responsive the delay due to join() may be substantial.
Before the change the test grepped the coordinator's HTTP request log through docker exec, which needed request logging enabled and a log-flush retry loop. Record HEAD requests on the client's own `requests.Session` instead. This removes the only caller of get_trino_container(), so delete it too.
The client made no requests between fetch() calls. A caller that held a batch of rows for longer than `query.client.timeout` got the query abandoned by the coordinator. The existing heartbeat ran only during spooled segment downloads, on a thread that shared the not-thread-safe `requests.Session` with regular traffic. Mirror the Java client instead. Check the clock for every served row. Once a full `heartbeat_interval` passes without a request, send one HEAD to the current `nextUri` on the caller's own thread. This renews the coordinator's abandonment timer without a background thread. An actively fetching query sends no extra traffic. Servers without HEAD support keep the old abandonment behavior.
Before the change Connection.close() closed the HTTP session and left running queries to hit `query.client.timeout` on the coordinator. Close the connection's cursors first; closing a cursor cancels its running query.
fb4e162 to
99c40ee
Compare
|
I improved the heartbeating by stealing from the JDBC statement client design and adapting it for Python. The only difference is sync IO instead of async (which requires a much bigger surgery and switching libraries). The blocking is bounded though and no thread safety issues. Thanks for asking the question @azawlocki-sbdt, made me wonder if we can do it without threads in the first place. |
| if isinstance(self._pending_segment.segment, SpooledSegment) and self._request and self._heartbeat_interval: | ||
| # Downloading a spooled segment may take some time. In the meantime, send heartbeat | ||
| # requests so the coordinator doesn't think we lost interest and close the query. | ||
| with _RequestHeartbeat(self._request, self._heartbeat_interval): | ||
| rows = self._decoder.decode(self._pending_segment.segment) | ||
| else: | ||
| rows = self._decoder.decode(self._pending_segment.segment) | ||
| rows = self._decoder.decode(self._pending_segment.segment) |
There was a problem hiding this comment.
As far as I remember, the whole point of _RequestHeartbeat is to send heartbeat requests asynchronously while the main thread is blocked on self._decoder.decode(). Are you sure it's ok to turn it off?
3 participants
Description
Fixes #463.
The fix mirrors the Java client's
StatementClientV1.TrinoResultruns a heartbeat check for every row it serves. Once a fullheartbeat_intervalpasses without a request the client sends one HEAD to the query's currentnextUriwhich resets the coordinator's abandonment timer. Every processed response defers the next beat, so an actively fetching query sends no extra requests.There is no background thread. The HEAD goes out on the caller's own thread so it never races other traffic on the
requests.Sessionwhich is not thread safe. The HEAD caps its timeouts at 5 seconds which bounds the pause it can add to a row loop.Servers older than Trino 475 reject HEAD on the statement endpoint and the client falls back to the old behavior there.
Additionally
Connection.close()now closes its cursors and cancels their running queries.Non-technical explanation
The client pings the coordinator between fetches, so slow consumers no longer fail with "Query was abandoned by the client".
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: